feat: handle CloudEvent and Message responses from function invocation

We already **kind of** handled them, but these changes make it more intuitive
for the user. For example, the function author can respond with `new CloudEvent(...)`
or `HTTP.binary(event)` and the framework should handle the response correctly.

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

Author: lance

lib/context.js

@@ -1,4 +1,4 @@
-const Spec = require('../lib/ce-constants.js').Spec;
+const { CloudEvent } = require('cloudevents');
 
 class Context {
   constructor(request) {
@@ -24,34 +24,31 @@ class CloudEventResponse {
   #response;
 
   constructor(response) {
-    this.#response = response;
-    if (!this.#response.headers) {
-      this.#response.headers = [];
-    }
+    this.#response = { data: response };
   }
 
   version(version) {
-    this.#response.headers[Spec.version] = version;
+    this.#response.specversion = version;
     return this;
   }
 
   id(id) {
-    this.#response.headers[Spec.id] = id;
+    this.#response.id = id;
     return this;
   }
 
   type(type) {
-    this.#response.headers[Spec.type] = type;
+    this.#response.type = type;
     return this;
   }
 
   source(source) {
-    this.#response.headers[Spec.source] = source;
+    this.#response.source = source;
     return this;
   }
 
   response() {
-    return this.#response;
+    return new CloudEvent(this.#response);
   }
 }
 

lib/invoker.js

@@ -1,4 +1,5 @@
 'use strict';
+const { CloudEvent, HTTP } = require('cloudevents');
 
 module.exports = function invoker(func) {
   return async function invokeFunction(context, log) {
@@ -7,10 +8,10 @@ module.exports = function invoker(func) {
       code: 200,
       response: undefined,
       headers: {
-        'Content-Type': 'application/json; charset=utf8',
-        'Access-Control-Allow-Methods':
+        'content-type': 'application/json; charset=utf8',
+        'access-control-allow-methods':
           'OPTIONS, GET, DELETE, POST, PUT, HEAD, PATCH',
-        'Access-Control-Allow-Origin': '*'
+        'access-control-allow-origin': '*'
       }
     };
 
@@ -47,9 +48,32 @@ module.exports = function invoker(func) {
 
     // Check for user defined headers
     if (typeof payload.response.headers === 'object') {
-      Object.assign(payload.headers, payload.response.headers);
+      const headers = {};
+      // normalize the headers as lowercase
+      for (const header in payload.response.headers) {
+        headers[header.toLocaleLowerCase()] = payload.response.headers[header];
+      }
+      payload.headers = { ...payload.headers, ...headers };
       delete payload.response.headers;
     }
+
+    // If the response is a CloudEvent, we need to convert it
+    // to a Message first and respond with the headers/body
+    if (payload.response instanceof CloudEvent) {
+      try {
+        const message = HTTP.binary(payload.response);
+        payload.headers = {...payload.headers, ...message.headers};
+        payload.response = message.body;  
+      } catch (e) {
+        console.error(e);
+      }
+    }
+
+    // Check for user supplied body
+    if (payload.response.body !== undefined) {
+      payload.response = payload.response.body;
+      delete payload.response.body;
+    }
     return payload;
   };
 };

lib/request-handler.js

@@ -22,7 +22,8 @@ module.exports = function(fastify, opts, done) {
 };
 
 function sendReply(reply, payload) {
-  if (payload.headers['Content-Type'].startsWith('text/plain')) {
+  const contentType = payload.headers['content-type'];
+  if (contentType.startsWith('text/plain') && (typeof payload.response !== 'string')) {
     payload.response = JSON.stringify(payload.response);
   }
   return reply

package-lock.json

@@ -21,7 +21,7 @@
   "bin": "./bin/cli.js",
   "dependencies": {
     "chalk": "^4.1.0",
-    "cloudevents": "^3.1.0",
+    "cloudevents": "^3.2.0",
     "commander": "^6.1.0",
     "death": "^1.1.0",
     "fastify": "^3.3.0",

package.json

@@ -1,3 +1,3 @@
 module.exports = function testFunc(context) {
-  return context;
+  return { ...context.query };
 };

test/fixtures/query-params/index.js

@@ -43,8 +43,8 @@ test('Provides HTTP request query parameters with the context parameter', t => {
       .expect('Content-Type', /json/)
       .end((err, res) => {
         t.error(err, 'No error');
-        t.equal(res.body.query.lunch, 'tacos');
-        t.equal(res.body.query.supper, 'burgers');
+        t.equal(res.body.lunch, 'tacos');
+        t.equal(res.body.supper, 'burgers');
         t.end();
         server.close();
       });

test/test-context.js

@@ -9,6 +9,7 @@ const Spec = require('../lib/ce-constants.js').Spec;
 const { existsSync, readdirSync } = require('fs');
 const { execSync } = require('child_process');
 const path = require('path');
+const { CloudEvent, HTTP } = require('cloudevents');
 
 // Ensure fixture dependencies are installed
 const fixtureDir = path.join(__dirname, 'fixtures');
@@ -172,6 +173,62 @@ test('Responds to 1.0 structured cloud events', t => {
   }, { log: false });
 });
 
+test('Handles 1.0 CloudEvent responses', t => {
+  framework(_ => {
+    return new CloudEvent({
+      source: 'test',
+      type: 'test-type',
+      data: 'some data',
+      datacontenttype: 'text/plain'
+    });
+  }, 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, '1.0')
+    .expect(200)
+    .expect('Content-Type', /text/)
+    .end((err, res) => {
+      t.error(err, 'No error');
+      t.equal(res.text, 'some data');
+      t.end();
+      server.close();
+    });
+  },
+  { log: false });
+});
+
+test('Handles 1.0 CloudEvent Message responses', t => {
+  framework(_ => {
+    return HTTP.binary(new CloudEvent({
+      source: 'test',
+      type: 'test-type',
+      data: 'some data',
+      datacontenttype: 'text/plain'
+    }));
+  }, 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, '1.0')
+    .expect(200)
+    .expect('Content-Type', /text/)
+    .end((err, res) => {
+      t.error(err, 'No error');
+      t.equal(res.text, 'some data');
+      t.end();
+      server.close();
+    });
+  },
+  { log: false });
+});
+
 test('Extracts event data as the first parameter to a function', t => {
   const data = {
     lunch: "tacos"
@@ -202,25 +259,6 @@ test('Extracts event data as the first parameter to a function', t => {
   }, { log: false });
 });
 
-test('Responds with error code (4xx or 5xx) to malformed 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.version, '0.3')
-      .set('ce-datacontenttype', 'application/json')
-      .expect('Content-Type', /json/)
-      .end((err, res) => {
-        t.assert(res.statusCode >= 400 && res.statusCode <= 599, 'Error code 4xx or 5xx expected.');
-        t.end();
-        server.close();
-      });
-  }, { log: false });
-});
-
 test('Responds with 406 Not Acceptable to unknown cloud event versions', t => {
   const func = require(`${__dirname}/fixtures/cloud-event/`);
   framework(func, server => {