src: remove HTTP receiver implementations and use messages/HTTP

This commit removes the binary|structured receiver implementations in
transport/http and uses instead the HTTP Message implementation.

The concept of a sender/invoker is really just sugar coating and should
probably be left up to the user to send an HTTP Message however they
choose. To be more consistent with the 'binary' and 'structured'
Message serialization functions, the 'Receiver' interface has been
changed to 'Deserializer', and the exported function named 'toEvent()'.

test(conformance): use Message and HTTP.toEvent() for conformance tests
src: export types from src/messages at top level

Signed-off-by: Lance Ball <[email protected]>

Author: lance

src/index.ts

@@ -3,9 +3,9 @@ import { ValidationError } from "./event/validation";
 import { CloudEventV03, CloudEventV03Attributes, CloudEventV1, CloudEventV1Attributes } from "./event/interfaces";
 
 import { Emitter, TransportOptions } from "./transport/emitter";
-import { Receiver, Mode } from "./transport/receiver";
+import { Receiver } from "./transport/receiver";
 import { Protocol } from "./transport/protocols";
-import { Headers } from "./messages";
+import { Headers, Mode, Binding, HTTP, Message, Serializer, Deserializer } from "./messages";
 
 import CONSTANTS from "./constants";
 
@@ -18,13 +18,19 @@ export {
   CloudEventV1Attributes,
   Version,
   ValidationError,
+  // From messages
+  Headers,
+  Mode,
+  Binding,
+  Message,
+  Deserializer,
+  Serializer,
+  HTTP,
   // From transport
   Emitter,
   Receiver,
-  Mode,
   Protocol,
   TransportOptions,
-  Headers,
   // From Constants
   CONSTANTS,
 };

src/messages/http/index.ts

@@ -1,17 +1,10 @@
-import { CloudEvent, CloudEventV03, CloudEventV1, CONSTANTS, Version } from "../..";
-import { Message, Sender, Headers } from "..";
+import { CloudEvent, CloudEventV03, CloudEventV1, CONSTANTS, Mode, Version } from "../..";
+import { Message, Headers } from "..";
 
-import { headersFor, v1binaryParsers, validate } from "./headers";
+import { headersFor, sanitize, v03structuredParsers, v1binaryParsers, v1structuredParsers, validate } from "./headers";
 import { asData, isBase64, isString, isStringOrObjectOrThrow, ValidationError } from "../../event/validation";
 import { validateCloudEvent } from "../../event/spec";
-import { MappedParser, parserByContentType } from "../../parsers";
-
-// Sender is a function that takes headers and body for transmission
-// over HTTP. Users supply this function as a parameter to HTTP.emit()
-// Sends a message by invoking sender(). Implements Invoker
-export function invoke(sender: Sender, message: Message): Promise<boolean> {
-  return sender(message.headers, message.body);
-}
+import { Base64Parser, JSONParser, MappedParser, Parser, parserByContentType } from "../../parsers";
 
 // implements Serializer
 export function binary(event: CloudEvent): Message {
@@ -33,6 +26,69 @@ export function structured(event: CloudEvent): Message {
   };
 }
 
+/**
+ * Converts a Message to a CloudEvent
+ *
+ * @param {Message} message the incoming message
+ * @return {CloudEvent} A new {CloudEvent} instance
+ */
+export function deserialize(message: Message): CloudEvent {
+  const cleanHeaders: Headers = sanitize(message.headers);
+  const mode: Mode = getMode(cleanHeaders);
+  let version = getVersion(mode, cleanHeaders, message.body);
+  if (version !== Version.V03 && version !== Version.V1) {
+    console.error(`Unknown spec version ${version}. Default to ${Version.V1}`);
+    version = Version.V1;
+  }
+  switch (mode) {
+    case Mode.BINARY:
+      return parseBinary(message, version);
+    case Mode.STRUCTURED:
+      return parseStructured(message, version);
+    default:
+      throw new ValidationError("Unknown Message mode");
+  }
+}
+
+/**
+ * Determines the HTTP transport mode (binary or structured) based
+ * on the incoming HTTP headers.
+ * @param {Headers} headers the incoming HTTP headers
+ * @returns {Mode} the transport mode
+ */
+function getMode(headers: Headers): Mode {
+  const contentType = headers[CONSTANTS.HEADER_CONTENT_TYPE];
+  if (contentType && contentType.startsWith(CONSTANTS.MIME_CE)) {
+    return Mode.STRUCTURED;
+  }
+  if (headers[CONSTANTS.CE_HEADERS.ID]) {
+    return Mode.BINARY;
+  }
+  throw new ValidationError("no cloud event detected");
+}
+
+/**
+ * Determines the version of an incoming CloudEvent based on the
+ * HTTP headers or HTTP body, depending on transport mode.
+ * @param {Mode} mode the HTTP transport mode
+ * @param {Headers} headers the incoming HTTP headers
+ * @param {Record<string, unknown>} body the HTTP request body
+ * @returns {Version} the CloudEvent specification version
+ */
+function getVersion(mode: Mode, headers: Headers, body: string | Record<string, string>) {
+  if (mode === Mode.BINARY) {
+    // Check the headers for the version
+    const versionHeader = headers[CONSTANTS.CE_HEADERS.SPEC_VERSION];
+    if (versionHeader) {
+      return versionHeader;
+    }
+  } else {
+    // structured mode - the version is in the body
+    return typeof body === "string" ? JSON.parse(body).specversion : (body as CloudEvent).specversion;
+  }
+  return Version.V1;
+}
+
 /**
  * Parses an incoming HTTP Message, converting it to a {CloudEvent}
  * instance if it conforms to the Cloud Event specification for this receiver.
@@ -42,7 +98,7 @@ export function structured(event: CloudEvent): Message {
  * @returns {CloudEvent} an instance of CloudEvent representing the incoming request
  * @throws {ValidationError} of the event does not conform to the spec
  */
-export function receive(message: Message, version: Version = Version.V1): CloudEvent {
+function parseBinary(message: Message, version: Version): CloudEvent {
   const headers = message.headers;
   let body = message.body;
 
@@ -102,3 +158,72 @@ export function receive(message: Message, version: Version = Version.V1): CloudE
   validateCloudEvent(cloudevent);
   return cloudevent;
 }
+
+/**
+ * Creates a new CloudEvent instance based on the provided payload and headers.
+ *
+ * @param {Message} message the incoming Message
+ * @param {Version} version the spec version of this message (v1 or v03)
+ * @returns {CloudEvent} a new CloudEvent instance for the provided headers and payload
+ * @throws {ValidationError} if the payload and header combination do not conform to the spec
+ */
+function parseStructured(message: Message, version: Version): CloudEvent {
+  let payload = message.body;
+  const headers = message.headers;
+
+  if (!payload) throw new ValidationError("payload is null or undefined");
+  if (!headers) throw new ValidationError("headers is null or undefined");
+  isStringOrObjectOrThrow(payload, new ValidationError("payload must be an object or a string"));
+
+  if (
+    headers[CONSTANTS.CE_HEADERS.SPEC_VERSION] &&
+    headers[CONSTANTS.CE_HEADERS.SPEC_VERSION] != Version.V03 &&
+    headers[CONSTANTS.CE_HEADERS.SPEC_VERSION] != Version.V1
+  ) {
+    throw new ValidationError(`invalid spec version ${headers[CONSTANTS.CE_HEADERS.SPEC_VERSION]}`);
+  }
+
+  payload = isString(payload) && isBase64(payload) ? Buffer.from(payload as string, "base64").toString() : payload;
+
+  // Clone and low case all headers names
+  const sanitizedHeaders = sanitize(headers);
+
+  const contentType = sanitizedHeaders[CONSTANTS.HEADER_CONTENT_TYPE];
+  const parser: Parser = contentType ? parserByContentType[contentType] : new JSONParser();
+  if (!parser) throw new ValidationError(`invalid content type ${sanitizedHeaders[CONSTANTS.HEADER_CONTENT_TYPE]}`);
+  const incoming = { ...(parser.parse(payload) as Record<string, unknown>) };
+
+  const eventObj: { [key: string]: unknown } = {};
+  const parserMap: Record<string, MappedParser> = version === Version.V1 ? v1structuredParsers : v03structuredParsers;
+
+  for (const key in parserMap) {
+    const property = incoming[key];
+    if (property) {
+      const parser: MappedParser = parserMap[key];
+      eventObj[parser.name] = parser.parser.parse(property as string);
+    }
+    delete incoming[key];
+  }
+
+  // extensions are what we have left after processing all other properties
+  for (const key in incoming) {
+    eventObj[key] = incoming[key];
+  }
+
+  // ensure data content is correctly encoded
+  if (eventObj.data && eventObj.datacontentencoding) {
+    if (eventObj.datacontentencoding === CONSTANTS.ENCODING_BASE64 && !isBase64(eventObj.data)) {
+      throw new ValidationError("invalid payload");
+    } else if (eventObj.datacontentencoding === CONSTANTS.ENCODING_BASE64) {
+      const dataParser = new Base64Parser();
+      eventObj.data = JSON.parse(dataParser.parse(eventObj.data as string));
+      delete eventObj.datacontentencoding;
+    }
+  }
+
+  const cloudevent = new CloudEvent(eventObj as CloudEventV1 | CloudEventV03);
+
+  // Validates the event
+  validateCloudEvent(cloudevent);
+  return cloudevent;
+}

src/messages/index.ts

@@ -1,5 +1,5 @@
-import { CloudEvent, Version } from "..";
-import { binary, invoke, receive, structured } from "./http";
+import { CloudEvent } from "..";
+import { binary, deserialize, structured } from "./http";
 
 /**
  * Binding is an interface for transport protocols to implement,
@@ -9,8 +9,7 @@ import { binary, invoke, receive, structured } from "./http";
 export interface Binding {
   binary: Serializer;
   structured: Serializer;
-  send: Invoker;
-  receive: Receiver;
+  toEvent: Deserializer;
 }
 
 /**
@@ -31,43 +30,32 @@ export interface Message {
 }
 
 /**
- * Serializer is an interface for functions that can convert a
- * CloudEvent into a Message.
- */
-export interface Serializer {
-  (event: CloudEvent): Message;
-}
-
-/**
- * Sender is a function interface for user-supplied functions
- * capable of transmitting a Message over a specific protocol
+ * An enum representing the two transport modes, binary and structured
  */
-export interface Sender {
-  (headers: Headers, body: string): Promise<boolean>;
+export enum Mode {
+  BINARY = "binary",
+  STRUCTURED = "structured",
 }
 
 /**
- * Invoker is an interface for functions that send a message
- * over a specific protocol by invoking the user-supplied
- * Sender function with the message headers and body.
- * TODO: This might be overkill
+ * Serializer is an interface for functions that can convert a
+ * CloudEvent into a Message.
  */
-export interface Invoker {
-  (sender: Sender, message: Message): Promise<boolean>;
+export interface Serializer {
+  (event: CloudEvent): Message;
 }
 
 /**
- * Receiver is a function interface that converts an incoming
+ * Deserializer is a function interface that converts a
  * Message to a CloudEvent
  */
-export interface Receiver {
-  (message: Message, version: Version | undefined): CloudEvent;
+export interface Deserializer {
+  (message: Message): CloudEvent;
 }
 
-// Export HTTP transport capabilities
+// HTTP Message capabilities
 export const HTTP: Binding = {
   binary: binary as Serializer,
   structured: structured as Serializer,
-  send: invoke as Invoker,
-  receive: receive as Receiver,
+  toEvent: deserialize as Deserializer,
 };