feat!: add http transport and remove axios

package.json

@@ -127,7 +127,7 @@
     "@typescript-eslint/parser": "^4.29.0",
     "ajv-cli": "^5.0.0",
     "ajv-formats": "^2.1.1",
-    "axios": "^0.21.3",
+    "axios": "^0.26.1",
     "chai": "~4.2.0",
     "eslint": "^7.32.0",
     "eslint-config-standard": "^16.0.3",

src/index.ts

@@ -8,6 +8,7 @@ import { ValidationError } from "./event/validation";
 import { CloudEventV1, CloudEventV1Attributes } from "./event/interfaces";
 
 import { Options, TransportFunction, EmitterFunction, emitterFor, Emitter } from "./transport/emitter";
+import { httpTransport } from "./transport/http";
 import { 
   Headers, Mode, Binding, HTTP, Kafka, KafkaEvent, KafkaMessage, Message, MQTT, MQTTMessage, MQTTMessageFactory,
   Serializer, Deserializer } from "./message";
@@ -25,6 +26,7 @@ export {
   MQTT,
   MQTTMessageFactory,
   emitterFor,
+  httpTransport,
   Emitter,
   // From Constants
   CONSTANTS

src/transport/http/index.ts

@@ -3,20 +3,61 @@
  SPDX-License-Identifier: Apache-2.0
 */
 
+import { Socket } from "net";
+import http, { OutgoingHttpHeaders } from "http";
+import https, { RequestOptions } from "https";
+
 import { Message, Options } from "../..";
-import axios from "axios";
+import { TransportFunction } from "../emitter";
+
+/**
+ * httpTransport provides a simple HTTP Transport function, which can send a CloudEvent,
+ * encoded as a Message to the endpoint. The returned function can be used with emitterFor()
+ * to provide an event emitter, for example:
+ * 
+ * const emitter = emitterFor(httpTransport("http://example.com"));
+ * emitter.emit(myCloudEvent)
+ *    .then(resp => console.log(resp));
+ * 
+ * @param {string|URL} sink the destination endpoint for the event
+ * @returns {TransportFunction} a function which can be used to send CloudEvents to _sink_
+ */
+export function httpTransport(sink: string | URL): TransportFunction {
+  const url = new URL(sink);
+  let base: any;
+  if (url.protocol === "https:") {
+    base = https;
+  } else if (url.protocol === "http:") {
+    base = http;
+  } else {
+    throw new TypeError(`unsupported protocol ${url.protocol}`);
+  }
+  return function(message: Message, options?: Options): Promise<unknown> {
+    return new Promise((resolve, reject) => {
+      options = { ...options };
 
-export function axiosEmitter(sink: string) {
-  return function (message: Message, options?: Options): Promise<unknown> {
-    options = { ...options };
-    const headers = {
-      ...message.headers,
-      ...(options.headers as Record<string, string>),
-    };
-    delete options.headers;
-    return axios.post(sink, message.body, {
-      headers: headers,
-      ...options,
+      // TODO: Callers should be able to set any Node.js RequestOptions
+      const opts: RequestOptions = {
+        method: "POST",
+        headers: {...message.headers, ...options.headers as OutgoingHttpHeaders},
+      };
+      try {
+        const response = {
+          body: "",
+          headers: {},
+        };
+        const req = base.request(url, opts, (res: Socket) => {
+          res.setEncoding("utf-8");
+          response.headers = (res as any).headers;
+          res.on("data", (chunk) => response.body += chunk);
+          res.on("end", () => { resolve(response); });
+        });
+        req.on("error", reject);
+        req.write(message.body);
+        req.end();
+      } catch (err) {
+        reject(err);
+      }
     });
   };
 }

test/integration/emitter_factory_test.ts

@@ -6,12 +6,13 @@
 import "mocha";
 import { expect } from "chai";
 import nock from "nock";
-import axios from "axios";
+import axios, { AxiosRequestHeaders } from "axios";
 import request from "superagent";
 import got from "got";
 
 import CONSTANTS from "../../src/constants";
-import { CloudEvent, emitterFor, HTTP, Mode, Message, Options, TransportFunction } from "../../src";
+import { CloudEvent, HTTP, Message, Mode, Options, TransportFunction, emitterFor, httpTransport }
+  from "../../src";
 
 const DEFAULT_CE_CONTENT_TYPE = CONSTANTS.DEFAULT_CE_CONTENT_TYPE;
 const sink = "https://cloudevents.io/";
@@ -38,7 +39,7 @@ export const fixture = new CloudEvent({
 });
 
 function axiosEmitter(message: Message, options?: Options): Promise<unknown> {
-  return axios.post(sink, message.body, { headers: message.headers, ...options });
+  return axios.post(sink, message.body, { headers: message.headers as AxiosRequestHeaders, ...options });
 }
 
 function superagentEmitter(message: Message, options?: Options): Promise<unknown> {
@@ -83,7 +84,6 @@ describe("emitterFor() defaults", () => {
 
   it("Supports HTTP binding, structured mode", () => {
     function transport(message: Message): Promise<unknown> {
-      console.error(message);
       // A structured message will have the application/cloudevents+json header
       expect(message.headers["content-type"]).to.equal(CONSTANTS.DEFAULT_CE_CONTENT_TYPE);
       const body = JSON.parse(message.body as string);
@@ -101,33 +101,50 @@ describe("emitterFor() defaults", () => {
   });
 });
 
+function setupMock(uri: string) {
+  nock(uri)
+  .post("/")
+  .reply(function (uri: string, body: nock.Body) {
+    // return the request body and the headers so they can be
+    // examined in the test
+    if (typeof body === "string") {
+      body = JSON.parse(body);
+    }
+    const returnBody = { ...(body as Record<string, unknown>), ...this.req.headers };
+    return [201, returnBody];
+  });
+}
+
 describe("HTTP Transport Binding for emitterFactory", () => {
-  beforeEach(() => {
-    nock(sink)
-      .post("/")
-      .reply(function (uri: string, body: nock.Body) {
-        // return the request body and the headers so they can be
-        // examined in the test
-        if (typeof body === "string") {
-          body = JSON.parse(body);
-        }
-        const returnBody = { ...(body as Record<string, unknown>), ...this.req.headers };
-        return [201, returnBody];
-      });
+  beforeEach(() => { setupMock(sink); });
+
+  describe("HTTPS builtin", () => {
+    testEmitterBinary(httpTransport(sink), "body");
+  });
+
+  describe("HTTP builtin", () => {
+    setupMock("http://cloudevents.io");
+    testEmitterBinary(httpTransport("http://cloudevents.io"), "body");
+    setupMock("http://cloudevents.io");
+    testEmitterStructured(httpTransport("http://cloudevents.io"), "body");
   });
 
   describe("Axios", () => {
-    testEmitter(axiosEmitter, "data");
+    testEmitterBinary(axiosEmitter, "data");
+    testEmitterStructured(axiosEmitter, "data");
   });
   describe("SuperAgent", () => {
-    testEmitter(superagentEmitter, "body");
+    testEmitterBinary(superagentEmitter, "body");
+    testEmitterStructured(superagentEmitter, "body");
   });
+
   describe("Got", () => {
-    testEmitter(gotEmitter, "body");
+    testEmitterBinary(gotEmitter, "body");
+    testEmitterStructured(gotEmitter, "body");
   });
 });
 
-function testEmitter(fn: TransportFunction, bodyAttr: string) {
+function testEmitterBinary(fn: TransportFunction, bodyAttr: string) {
   it("Works as a binary event emitter", async () => {
     const emitter = emitterFor(fn);
     const response = (await emitter(fixture)) as Record<string, Record<string, string>>;
@@ -137,7 +154,9 @@ function testEmitter(fn: TransportFunction, bodyAttr: string) {
     }
     assertBinary(body);
   });
+}
 
+function testEmitterStructured(fn: TransportFunction, bodyAttr: string) {
   it("Works as a structured event emitter", async () => {
     const emitter = emitterFor(fn, { binding: HTTP, mode: Mode.STRUCTURED });
     const response = (await emitter(fixture)) as Record<string, Record<string, Record<string, string>>>;

webpack.config.js

@@ -6,7 +6,9 @@ module.exports = {
   },
   resolve: {
     fallback: {
-      util: require.resolve("util/")
+      util: require.resolve("util/"),
+      http: false,
+      https: false
     },
   },
   output: {