feat: add support for kafka transport

This commit extends the `message` package to include Kafka transport.

Additionally, some of the type information has changed across the project
to more accurately reflect the type of `Message` (by including `T`).

Related: #390

Author: lance

package-lock.json

@@ -25,7 +25,7 @@ import { JSONParser, MappedParser, Parser, parserByContentType } from "../../par
  * @param {CloudEvent} event The event to serialize
  * @returns {Message} a Message object with headers and body
  */
-export function binary<T>(event: CloudEvent<T>): Message {
+export function binary<T>(event: CloudEvent<T>): Message<T> {
   const contentType: Headers = { [CONSTANTS.HEADER_CONTENT_TYPE]: CONSTANTS.DEFAULT_CONTENT_TYPE };
   const headers: Headers = { ...contentType, ...headersFor(event) };
   let body = event.data;
@@ -47,7 +47,7 @@ export function binary<T>(event: CloudEvent<T>): Message {
  * @param {CloudEvent} event the CloudEvent to be serialized
  * @returns {Message} a Message object with headers and body
  */
-export function structured<T>(event: CloudEvent<T>): Message {
+export function structured<T>(event: CloudEvent<T>): Message<T> {
   if (event.data_base64) {
     // The event's data is binary - delete it
     event = event.cloneWith({ data: undefined });
@@ -67,7 +67,7 @@ export function structured<T>(event: CloudEvent<T>): Message {
  * @param {Message} message an incoming Message object
  * @returns {boolean} true if this Message is a CloudEvent
  */
-export function isEvent(message: Message): boolean {
+export function isEvent<T>(message: Message<T>): boolean {
   // TODO: this could probably be optimized
   try {
     deserialize(message);
@@ -84,7 +84,7 @@ export function isEvent(message: Message): boolean {
  * @param {Message} message the incoming message
  * @return {CloudEvent} A new {CloudEvent} instance
  */
-export function deserialize<T>(message: Message): CloudEvent<T> | CloudEvent<T>[] {
+export function deserialize<T>(message: Message<T>): CloudEvent<T> | CloudEvent<T>[] {
   const cleanHeaders: Headers = sanitize(message.headers);
   const mode: Mode = getMode(cleanHeaders);
   const version = getVersion(mode, cleanHeaders, message.body);
@@ -157,7 +157,7 @@ function getVersion(mode: Mode, headers: Headers, body: string | Record<string,
  * @returns {CloudEvent} an instance of CloudEvent representing the incoming request
  * @throws {ValidationError} of the event does not conform to the spec
  */
-function parseBinary<T>(message: Message, version: Version): CloudEvent<T> {
+function parseBinary<T>(message: Message<T>, version: Version): CloudEvent<T> {
   const headers = { ...message.headers };
   let body = message.body;
 
@@ -208,7 +208,7 @@ function parseBinary<T>(message: Message, version: Version): CloudEvent<T> {
  * @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<T>(message: Message, version: Version): CloudEvent<T> {
+function parseStructured<T>(message: Message<T>, version: Version): CloudEvent<T> {
   const payload = message.body;
   const headers = message.headers;
 
@@ -253,7 +253,7 @@ function parseStructured<T>(message: Message, version: Version): CloudEvent<T> {
   return new CloudEvent<T>(eventObj as CloudEventV1<T>, false);
 }
 
-function parseBatched<T>(message: Message): CloudEvent<T> | CloudEvent<T>[] {
+function parseBatched<T>(message: Message<T>): CloudEvent<T> | CloudEvent<T>[] {
   const ret: CloudEvent<T>[] = [];
   const events = JSON.parse(message.body as string);
   events.forEach((element: CloudEvent) => {

src/message/http/index.ts

@@ -41,9 +41,9 @@ export interface Headers extends IncomingHttpHeaders {
  * @property {@linkcode Headers} `headers` - the headers for the event Message
  * @property string `body` - the body of the event Message
  */
-export interface Message {
+export interface Message<T> {
   headers: Headers;
-  body: string | unknown;
+  body: T | string | Buffer | unknown;
 }
 
 /**
@@ -62,7 +62,7 @@ export enum Mode {
  * @interface
  */
 export interface Serializer {
-  <T>(event: CloudEvent<T>): Message;
+  <T>(event: CloudEvent<T>): Message<T>;
 }
 
 /**
@@ -71,7 +71,7 @@ export interface Serializer {
  * @interface
  */
 export interface Deserializer {
-  <T>(message: Message): CloudEvent<T> | CloudEvent<T>[];
+  <T>(message: Message<T>): CloudEvent<T> | CloudEvent<T>[];
 }
 
 /**
@@ -80,7 +80,7 @@ export interface Deserializer {
  * @interface
  */
 export interface Detector {
-  (message: Message): boolean;
+  <T>(message: Message<T>): boolean;
 }
 
 /**
@@ -93,3 +93,5 @@ export const HTTP: Binding = {
   toEvent: deserialize as Deserializer,
   isEvent: isEvent as Detector,
 };
+
+export * from "./kafka";

src/message/index.ts

@@ -0,0 +1,60 @@
+/*
+ Copyright 2021 The CloudEvents Authors
+ SPDX-License-Identifier: Apache-2.0
+*/
+
+import { CloudEvent, CONSTANTS, Headers } from "../..";
+
+/**
+ * The set of CloudEvent headers that may exist on a Kafka message
+ */
+export const KAFKA_CE_HEADERS: { [key: string]: string } = Object.freeze({
+  /** corresponds to the CloudEvent#id */
+  ID: "ce_id",
+  /** corresponds to the CloudEvent#type */
+  TYPE: "ce_type",
+  /** corresponds to the CloudEvent#source */
+  SOURCE: "ce_source",
+  /** corresponds to the CloudEvent#specversion */
+  SPEC_VERSION:  "ce_specversion",
+  /** corresponds to the CloudEvent#time */
+  TIME: "ce_time",
+  /** corresponds to the CloudEvent#subject */
+  SUBJECT: "ce_subject",
+  /** corresponds to the CloudEvent#datacontenttype */
+  DATACONTENTTYPE: "ce_datacontenttype",
+  /** corresponds to the CloudEvent#dataschema */
+  DATASCHEMA: "ce_dataschema",
+} as const);
+
+export const HEADER_MAP: { [key: string]: string } = {
+  [KAFKA_CE_HEADERS.ID]: "id",
+  [KAFKA_CE_HEADERS.TYPE]: "type",
+  [KAFKA_CE_HEADERS.SOURCE]: "source",
+  [KAFKA_CE_HEADERS.SPEC_VERSION]: "specversion",
+  [KAFKA_CE_HEADERS.TIME]: "time",
+  [KAFKA_CE_HEADERS.SUBJECT]: "subject",
+  [KAFKA_CE_HEADERS.DATACONTENTTYPE]: "datacontenttype",
+  [KAFKA_CE_HEADERS.DATASCHEMA]: "dataschema"
+};
+
+/**
+ * A conveninece function to convert a CloudEvent into headers
+ * @param {CloudEvent} event a CloudEvent object
+ * @returns {Headers} the CloudEvent attribute as Kafka headers
+ */
+export function headersFor<T>(event: CloudEvent<T>): Headers {
+  const headers: Headers = {};
+
+  Object.getOwnPropertyNames(event).forEach((property) => {
+    // Ignore the 'data' property
+    // it becomes the Kafka message's 'value' field
+    if (property != CONSTANTS.CE_ATTRIBUTES.DATA && property != CONSTANTS.STRUCTURED_ATTRS_1.DATA_BASE64) {
+      // all CloudEvent property names get prefixed with 'ce_'
+      // https://github.com/cloudevents/spec/blob/v1.0.1/kafka-protocol-binding.md#3231-property-names
+      headers[`ce_${property}`] = event[property] as string;
+    }
+  });
+
+  return headers;
+}