feat: introduce Message, Serializer, Deserializer and Binding interfaces (cloudevents#324)

* lib(messages): Implement a 4.0 Messages and other supporting interfaces

This commit introduces the Message, Serializer and Deserializer, and Binding
interfaces used to convert a CloudEvent into a Message that can be sent across
a transport protocol. The first protocol implemented for this is HTTP, and some
of the functionality formerly in src/transport/http has been simplified,
reduced and/or moved to /src/messages/http.

Test for V1 and V3 events are in place. Conformance tests have been modified to use
these new interfaces vs. the HTTP Receiver class.

Signed-off-by: Lance Ball <lball@redhat.com>

Author: lance

examples/express-ex/index.js

@@ -1,8 +1,7 @@
-/* eslint-disable no-console */
+/* eslint-disable */
 
 const express = require("express");
-const {Receiver} = require("cloudevents");
-
+const { Receiver } = require("cloudevents");
 const app = express();
 
 app.use((req, res, next) => {
@@ -25,8 +24,16 @@ app.post("/", (req, res) => {
 
   try {
     const event = Receiver.accept(req.headers, req.body);
-    console.log(`Accepted event: ${event}`);
-    res.status(201).json(event);
+    // respond as an event
+    const responseEventMessage = new CloudEvent({
+      source: '/',
+      type: 'event:response',
+      ...event
+    });
+    responseEventMessage.data = {
+      hello: 'world'
+    };
+    res.status(201).json(responseEventMessage);
   } catch (err) {
     console.error(err);
     res.status(415).header("Content-Type", "application/json").send(JSON.stringify(err));

examples/express-ex/package.json

@@ -14,7 +14,7 @@
   "author": "fabiojose@gmail.com",
   "license": "Apache-2.0",
   "dependencies": {
-    "cloudevents": "~3.0.0",
+    "cloudevents": "^3.1.0",
     "express": "^4.17.1"
   }
 }

src/event/cloudevent.ts

@@ -156,6 +156,12 @@ export class CloudEvent implements CloudEventV1, CloudEventV03 {
     this.#_data = value;
   }
 
+  /**
+   * Used by JSON.stringify(). The name is confusing, but this method is called by
+   * JSON.stringify() when converting this object to JSON.
+   * @see https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/JSON/stringify
+   * @return {object} this event as a plain object
+   */
   toJSON(): Record<string, unknown> {
     const event = { ...this };
     event.time = this.time;

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, headersFor } from "./transport/http/headers";
+import { Headers, Mode, Binding, HTTP, Message, Serializer, Deserializer, headersFor } from "./message";
 
 import CONSTANTS from "./constants";
 
@@ -18,14 +18,20 @@ export {
   CloudEventV1Attributes,
   Version,
   ValidationError,
-  // From transport
-  Emitter,
-  Receiver,
-  Mode,
-  Protocol,
-  TransportOptions,
+  // From message
   Headers,
-  headersFor,
+  Mode,
+  Binding,
+  Message,
+  Deserializer,
+  Serializer,
+  headersFor, // TODO: Deprecated. Remove for 4.0
+  HTTP,
+  // From transport
+  Emitter, // TODO: Deprecated. Remove for 4.0
+  Receiver, // TODO: Deprecated. Remove for 4.0
+  Protocol, // TODO: Deprecated. Remove for 4.0
+  TransportOptions, // TODO: Deprecated. Remove for 4.0
   // From Constants
   CONSTANTS,
 };

src/transport/http/versions.ts renamed to src/message/http/headers.ts

@@ -1,6 +1,97 @@
 import { PassThroughParser, DateParser, MappedParser } from "../../parsers";
+import { ValidationError, CloudEvent } from "../..";
+import { Headers } from "../";
+import { Version } from "../../event/cloudevent";
 import CONSTANTS from "../../constants";
 
+export const allowedContentTypes = [CONSTANTS.DEFAULT_CONTENT_TYPE, CONSTANTS.MIME_JSON, CONSTANTS.MIME_OCTET_STREAM];
+export const requiredHeaders = [
+  CONSTANTS.CE_HEADERS.ID,
+  CONSTANTS.CE_HEADERS.SOURCE,
+  CONSTANTS.CE_HEADERS.TYPE,
+  CONSTANTS.CE_HEADERS.SPEC_VERSION,
+];
+
+/**
+ * Validates cloud event headers and their values
+ * @param {Headers} headers event transport headers for validation
+ * @throws {ValidationError} if the headers are invalid
+ * @return {boolean} true if headers are valid
+ */
+export function validate(headers: Headers): Headers {
+  const sanitizedHeaders = sanitize(headers);
+
+  // if content-type exists, be sure it's an allowed type
+  const contentTypeHeader = sanitizedHeaders[CONSTANTS.HEADER_CONTENT_TYPE];
+  const noContentType = !allowedContentTypes.includes(contentTypeHeader);
+  if (contentTypeHeader && noContentType) {
+    throw new ValidationError("invalid content type", [sanitizedHeaders[CONSTANTS.HEADER_CONTENT_TYPE]]);
+  }
+
+  requiredHeaders
+    .filter((required: string) => !sanitizedHeaders[required])
+    .forEach((required: string) => {
+      throw new ValidationError(`header '${required}' not found`);
+    });
+
+  if (!sanitizedHeaders[CONSTANTS.HEADER_CONTENT_TYPE]) {
+    sanitizedHeaders[CONSTANTS.HEADER_CONTENT_TYPE] = CONSTANTS.MIME_JSON;
+  }
+
+  return sanitizedHeaders;
+}
+
+/**
+ * Returns the HTTP headers that will be sent for this event when the HTTP transmission
+ * mode is "binary". Events sent over HTTP in structured mode only have a single CE header
+ * and that is "ce-id", corresponding to the event ID.
+ * @param {CloudEvent} event a CloudEvent
+ * @returns {Object} the headers that will be sent for the event
+ */
+export function headersFor(event: CloudEvent): Headers {
+  const headers: Headers = {};
+  let headerMap: Readonly<{ [key: string]: MappedParser }>;
+  if (event.specversion === Version.V1) {
+    headerMap = v1headerMap;
+  } else {
+    headerMap = v03headerMap;
+  }
+
+  // iterate over the event properties - generate a header for each
+  Object.getOwnPropertyNames(event).forEach((property) => {
+    const value = event[property];
+    if (value) {
+      const map: MappedParser | undefined = headerMap[property] as MappedParser;
+      if (map) {
+        headers[map.name] = map.parser.parse(value as string) as string;
+      } else if (property !== CONSTANTS.DATA_ATTRIBUTE && property !== `${CONSTANTS.DATA_ATTRIBUTE}_base64`) {
+        headers[`${CONSTANTS.EXTENSIONS_PREFIX}${property}`] = value as string;
+      }
+    }
+  });
+  // Treat time specially, since it's handled with getters and setters in CloudEvent
+  if (event.time) {
+    headers[CONSTANTS.CE_HEADERS.TIME] = event.time as string;
+  }
+  return headers;
+}
+
+/**
+ * Sanitizes incoming headers by lowercasing them and potentially removing
+ * encoding from the content-type header.
+ * @param {Headers} headers HTTP headers as key/value pairs
+ * @returns {Headers} the sanitized headers
+ */
+export function sanitize(headers: Headers): Headers {
+  const sanitized: Headers = {};
+
+  Array.from(Object.keys(headers))
+    .filter((header) => Object.hasOwnProperty.call(headers, header))
+    .forEach((header) => (sanitized[header.toLowerCase()] = headers[header]));
+
+  return sanitized;
+}
+
 function parser(name: string, parser = new PassThroughParser()): MappedParser {
   return { name: name, parser: parser };
 }