chore: clean up event unmarshalling, remove async API calls

The cloudevents sdk has some inconsistencies with how it exposed the
unmarshalling. Version 0.3 and earlier had an async API, while 1.0
(smartly) changed that. As a result, the impl in this file was pretty
ugly. By using the receivers directly, we can avoid that and clean up
this code.

These changes are based on: cloudevents/sdk-javascript#120

Once those changes land in a released version of the cloudevents sdk
this file can be even simpler as the sdk will handle most of this work.

As an aside - dropped support for 0.2. Normally I wouldn't be so casual about
dropping support, but upstream sdk has no 0.2 support in the next release
and cloud events spec says only the two most recent versions are required
to be supported.

Author: lance

lib/event-handler.js

@@ -1,87 +1,95 @@
 'use strict';
 
-const v02 = require('cloudevents-sdk/v02');
-const v03 = require('cloudevents-sdk/v03');
-const v1 = require('cloudevents-sdk/v1');
+const V03Binary = require('cloudevents-sdk/lib/bindings/http/receiver_binary_0_3');
+const V03Structured = require('cloudevents-sdk/lib/bindings/http/receiver_structured_0_3.js');
+const V1Binary = require('cloudevents-sdk/lib/bindings/http/receiver_binary_1.js');
+const V1Structured = require('cloudevents-sdk/lib/bindings/http/receiver_structured_1.js');
+
 const Spec = require('./ce-constants.js').Spec;
 
-const v02Unmarshaller = new v02.HTTPUnmarshaller();
-const v03Unmarshaller = new v03.HTTPUnmarshaller();
-const v1BinaryReceiver = new v1.BinaryHTTPReceiver();
-const v1StructuredReceiver = new v1.StructuredHTTPReceiver();
+const receivers = {
+  v1: {
+    structured: new V1Structured(),
+    binary: new V1Binary()
+  },
+  v03: {
+    structured: new V03Structured(),
+    binary: new V03Binary()
+  }
+};
 
 function use(fastify, opts, done) {
   fastify.addContentTypeParser('application/cloudevents+json',
     { parseAs: 'string' },
     function(req, body, done) {
-      // unmarshallEvent() handles parsing
       done(null, body);
     });
 
   fastify.decorateRequest('isCloudEvent', function() {
     if (Spec.type in this.req.headers) {
       return true;
-    } else {
-      const contentType = this.req.headers['content-type'];
-      if (contentType && contentType.match(/application\/cloudevents/)) {
-        return true;
-      }
     }
-    return false;
+    const contentType = this.req.headers['content-type'];
+    return contentType && contentType.match(/application\/cloudevents/);
   });
 
   // Any incoming requests for cloud events will only be
   // processed if it's a cloud event spec version we know about
-  fastify.addHook('preHandler', async(request, reply) => {
+  fastify.addHook('preHandler', function(request, reply, done) {
     if (request.isCloudEvent()) {
-      const version = request.headers[Spec.version];
-      // if there is no version in the headers, it is a
-      // structured event
-      if (version && !acceptsVersion(version)) {
+      try {
+        request.context.cloudevent =
+          accept(request.headers, request.body).format();
+      } catch (err) {
         reply.code(406);
-        const error = new Error(
-          `Unsupported cloud event version detected: ${version}`);
-        error.code = 406;
-        throw error;
-      } else {
-        try {
-          await unmarshallEvent(request);
-        } catch (err) {
-          throw new Error(err.message ||
-            `Failed to unmarshall cloud event: ${err}`);
-        }
+        done(err);
       }
     }
+    done();
   });
 
   done();
 }
 
-async function unmarshallEvent(request) {
-  const version = request.headers[Spec.version];
-  if (!version) {
-    // it's a structured event and the version is in the body
-    // currently only v1 structured events are supported
-    try {
-      const event = v1StructuredReceiver.parse(request.body, request.headers);
-      request.context.cloudevent = event.format();
-    } catch (err) {
-      return Promise.reject(err);
+function accept(headers, body) {
+  const mode = getMode(headers);
+  const version = getVersion(mode, headers, body);
+  switch (version) {
+    case '1.0':
+      return receivers.v1[mode].parse(body, headers);
+    case '0.3':
+      return receivers.v03[mode].parse(body, headers);
+    default:
+      console.error(`Unknown spec version ${version}`);
+      throw new TypeError(
+        `Unsupported cloud event version detected: ${version}`);
+  }
+}
+
+function getMode(headers) {
+  let mode = 'binary';
+  if (headers['content-type']) {
+    if (headers['content-type'].startsWith('application/cloudevents')) {
+      mode = 'structured';
     }
-  } else if (version === '0.2') {
-    return v02Unmarshaller.unmarshall(request.body, request.headers)
-      .then(cloudevent => (request.context.cloudevent = cloudevent.format()));
-  } else if (version === '0.3') {
-    return v03Unmarshaller.unmarshall(request.body, request.headers)
-      .then(cloudevent => (request.context.cloudevent = cloudevent.format()));
-  } else if (version === '1.0') {
-    const event = v1BinaryReceiver.parse(request.body, request.headers);
-    request.context.cloudevent = event.format();
   }
+  return mode;
 }
 
-function acceptsVersion(version) {
-  return ['0.2', '0.3', '1.0'].find(elem => version === elem) !== undefined;
+function getVersion(mode, headers, body) {
+  let version = '1.0'; // default to 1.0
+
+  if (mode === 'binary') {
+    // Check the headers for the version
+    if (headers['ce-specversion']) {
+      version = headers['ce-specversion'];
+    }
+  } else {
+    // structured mode - the version is in the body
+    version = typeof body === 'string'
+      ? JSON.parse(body).specversion : body.specversion;
+  }
+  return version;
 }
 
 module.exports = exports = use;

test/test.js

@@ -79,27 +79,6 @@ test('Accepts HTTP POST requests', t => {
   }, { log: false });
 });
 
-test('Responds to 0.2 binary cloud events', t => {
-  const func = require(`${__dirname}/fixtures/cloud-event/`);
-  framework(func, server => {
-    request(server)
-      .post('/')
-      .send({ message: 'hello' })
-      .set(Spec.id, '1')
-      .set(Spec.source, 'integration-test')
-      .set(Spec.type, 'dev.knative.example')
-      .set(Spec.version, '0.2')
-      .expect(200)
-      .expect('Content-Type', /json/)
-      .end((err, res) => {
-        t.error(err, 'No error');
-        t.equal(res.body, 'hello');
-        t.end();
-        server.close();
-      });
-  }, { log: false });
-});
-
 test('Responds to 0.3 binary cloud events', t => {
   const func = require(`${__dirname}/fixtures/cloud-event/`);
   framework(func, server => {