✨ feat: enable HTTP trigger async function

Signed-off-by: Haili Zhang <haili.zhang@outlook.com>

Author: webup

src/function_wrappers.ts

@@ -14,6 +14,10 @@
 
 // eslint-disable-next-line node/no-deprecated-api
 import * as domain from 'domain';
+
+import * as Debug from 'debug';
+import {get, isEmpty} from 'lodash';
+
 import {Request, Response, RequestHandler} from 'express';
 import {sendCrashResponse} from './logger';
 import {sendResponse} from './invoker';
@@ -27,9 +31,14 @@ import {
   CloudEventFunctionWithCallback,
   HandlerFunction,
 } from './functions';
-import {CloudEvent} from './functions';
+import {CloudEvent, OpenFunction} from './functions';
 import {SignatureType} from './types';
 
+import {OpenFunctionContext} from './openfunction/function_context';
+import {OpenFunctionRuntime} from './openfunction/function_runtime';
+
+const debug = Debug('common:wrapper');
+
 /**
  * The handler function used to signal completion of event functions.
  */
@@ -122,6 +131,37 @@ const wrapHttpFunction = (execute: HttpFunction): RequestHandler => {
   };
 };
 
+const wrapHttpAsyncFunction = (
+  userFunction: OpenFunction,
+  context: OpenFunctionContext
+): RequestHandler => {
+  const ctx = OpenFunctionRuntime.ProxyContext(context);
+  const httpHandler = (req: Request, res: Response) => {
+    const callback = getOnDoneCallback(res);
+
+    Promise.resolve()
+      .then(() => userFunction(ctx, req.body))
+      .then(result => {
+        debug('ℹ️ User function returned: %j', result);
+
+        const data = get(result, 'body');
+        const code = get(result, 'code', 200);
+        const headers = get(result, 'headers');
+
+        !isEmpty(headers) && res.set(headers);
+
+        if (data !== undefined) {
+          res.status(code).send(data);
+        } else {
+          res.status(code).end();
+        }
+      })
+      .catch(err => callback(err, undefined));
+  };
+
+  return wrapHttpFunction(httpHandler);
+};
+
 /**
  * Wraps an async CloudEvent function in an express RequestHandler.
  * @param userFunction User's function.
@@ -202,10 +242,16 @@ const wrapEventFunctionWithCallback = (
  */
 export const wrapUserFunction = <T = unknown>(
   userFunction: HandlerFunction<T>,
-  signatureType: SignatureType
+  signatureType: SignatureType,
+  context?: object
 ): RequestHandler => {
   switch (signatureType) {
     case 'http':
+      if (!isEmpty((context as OpenFunctionContext)?.outputs))
+        return wrapHttpAsyncFunction(
+          userFunction as OpenFunction,
+          context as OpenFunctionContext
+        );
       return wrapHttpFunction(userFunction as HttpFunction);
     case 'event':
       // Callback style if user function has more than 2 arguments.

src/functions.ts

@@ -78,13 +78,38 @@ export interface CloudEventFunctionWithCallback<T = unknown> {
 }
 
 /**
- * A OpenFunction async function handler.
+ * An OpenFunction async function handler.
  * @public
  */
 export interface OpenFunction {
   (ctx: OpenFunctionRuntime, data: {}): any;
 }
 
+/**
+ * OpenFunction runtime as the context handler.
+ * @public
+ */
+export {OpenFunctionRuntime};
+
+/**
+ * HTTP response ouput from OpenFunction async function
+ * @public
+ */
+export interface HttpFunctionResponse {
+  /**
+   * Status code of the response.
+   */
+  code?: number;
+  /**
+   * Headers of the response.
+   */
+  headers?: Record<string, string>;
+  /**
+   * Body of the response.
+   */
+  body?: any;
+}
+
 /**
  * A function handler.
  * @public

src/openfunction/async_server.ts

@@ -1,4 +1,4 @@
-import {forEach, has, get} from 'lodash';
+import {forEach} from 'lodash';
 import {DaprServer} from 'dapr-client';
 
 import {OpenFunction} from '../functions';
@@ -20,7 +20,7 @@ export default function (
   context: OpenFunctionContext
 ): AsyncFunctionServer {
   const app = new DaprServer('localhost', context.port);
-  const ctx = getContextProxy(context);
+  const ctx = OpenFunctionRuntime.ProxyContext(context);
 
   const wrapper = async (data: object) => {
     await userFunction(ctx, data);
@@ -44,23 +44,3 @@ export default function (
 
   return app;
 }
-
-/**
- * It creates a proxy for the runtime object, which delegates all property access to the runtime object
- * @param {OpenFunctionContext} context - The context object to be proxied.
- * @returns The proxy object.
- */
-function getContextProxy(context: OpenFunctionContext): OpenFunctionRuntime {
-  // Get a proper runtime for the context
-  const runtime = OpenFunctionRuntime.Parse(context);
-
-  // Create a proxy for the context
-  return new Proxy(runtime, {
-    get: (target, prop) => {
-      // Provide delegated property access of the context object
-      if (has(target.context, prop)) return get(target.context, prop);
-      // Otherwise, return the property of the runtime object
-      else return Reflect.get(target, prop);
-    },
-  });
-}

src/openfunction/function_runtime.ts

@@ -1,8 +1,10 @@
 import {env} from 'process';
 
-import {chain} from 'lodash';
+import {chain, get, has} from 'lodash';
 import {DaprClient, CommunicationProtocolEnum} from 'dapr-client';
 
+import {HttpFunctionResponse} from '../functions';
+
 import {
   OpenFunctionComponent,
   OpenFunctionContext,
@@ -11,6 +13,7 @@ import {
 
 /**
  * The OpenFunction's serving runtime abstract class.
+ * @public
  */
 export abstract class OpenFunctionRuntime {
   /**
@@ -32,6 +35,26 @@ export abstract class OpenFunctionRuntime {
     return new DaprRuntime(context);
   }
 
+  /**
+   * It creates a proxy for the runtime object, which delegates all property access to the runtime object
+   * @param context - The context object to be proxied.
+   * @returns The proxy object.
+   */
+  static ProxyContext(context: OpenFunctionContext): OpenFunctionRuntime {
+    // Get a proper runtime for the context
+    const runtime = OpenFunctionRuntime.Parse(context);
+
+    // Create a proxy for the context
+    return new Proxy(runtime, {
+      get: (target, prop) => {
+        // Provide delegated property access of the context object
+        if (has(target.context, prop)) return get(target.context, prop);
+        // Otherwise, return the property of the runtime object
+        else return Reflect.get(target, prop);
+      },
+    });
+  }
+
   /**
    * Getter for the port of Dapr sidecar
    */
@@ -43,7 +66,26 @@ export abstract class OpenFunctionRuntime {
   }
 
   /**
-   * The promise that send data to certain ouput binding or pubsub topic.
+   * It returns an HTTP style response object with a `code`, `headers`, and `body` property
+   * @param body - The data you want to send back to the client.
+   * @param code - The HTTP status code to return.
+   * @param headers - An object containing the headers to be sent with the response.
+   * @returns A function that takes in data, code, and headers and returns an response object.
+   */
+  response(
+    body: unknown,
+    code = 200,
+    headers?: Record<string, string>
+  ): HttpFunctionResponse {
+    return {
+      code,
+      headers,
+      body,
+    };
+  }
+
+  /**
+   * The promise that send data to certain ouput.
    */
   abstract send(data: object, output?: string): Promise<object>;
 }

src/server.ts

@@ -24,13 +24,12 @@ import {cloudEventToBackgroundEventMiddleware} from './middleware/cloud_event_to
 import {backgroundEventToCloudEventMiddleware} from './middleware/background_event_to_cloud_event';
 import {wrapUserFunction} from './function_wrappers';
 
-import daprOutputMiddleware from './openfunction/dapr_output_middleware';
-
 /**
  * Creates and configures an Express application and returns an HTTP server
  * which will run it.
  * @param userFunction User's function.
  * @param functionSignatureType Type of user's function signature.
+ * @param context Optional context object.
  * @return HTTP server.
  */
 export function getServer(
@@ -53,9 +52,6 @@ export function getServer(
     next();
   });
 
-  // Use OpenFunction middlewares
-  app.use(daprOutputMiddleware);
-
   /**
    * Retains a reference to the raw body buffer to allow access to the raw body
    * for things like request signature validation.  This is used as the "verify"
@@ -139,7 +135,11 @@ export function getServer(
   }
 
   // Set up the routes for the user's function
-  const requestHandler = wrapUserFunction(userFunction, functionSignatureType);
+  const requestHandler = wrapUserFunction(
+    userFunction,
+    functionSignatureType,
+    context
+  );
   if (functionSignatureType === 'http') {
     app.all('/*', requestHandler);
   } else {

test/conformance/function.js

@@ -44,9 +44,10 @@ function writeJson(content) {
   fs.writeFileSync(fileName, json);
 }
 
-function tryKnative(req, res) {
-  debug('✅ Function should receive request: %o', req.body);
-  res.send(req.body);
+async function tryKnativeAsync(ctx, data) {
+  debug('✅ Function should receive request: %o', data);
+  await ctx.send(data);
+  return ctx.response(data);
 }
 
 function tryAsync(ctx, data) {
@@ -57,6 +58,6 @@ module.exports = {
   writeHttp,
   writeCloudEvent,
   writeLegacyEvent,
-  tryKnative,
+  tryKnativeAsync,
   tryAsync,
 };

test/conformance/package.json

@@ -8,10 +8,10 @@
   },
   "scripts": {
     "start": "functions-framework",
-    "knative": "concurrently npm:knative:run:* npm:knative:test",
-    "knative:run:func": "cross-env DEBUG=test:*,common:*,ofn:* env-cmd -e knative functions-framework --target=tryKnative",
-    "knative:run:dapr": "dapr run -H 3500 -d ../data/components/http --log-level warn",
-    "knative:test": "wait-on tcp:3500 tcp:8080 && curl -s -d '{\"data\": \"hello\"}' -H 'Content-Type: application/json' localhost:8080",
+    "knative:async": "concurrently npm:knative:async:run:* npm:knative:async:test",
+    "knative:async:run:func": "cross-env DEBUG=test:*,common:*,ofn:* env-cmd -e knative functions-framework --target=tryKnativeAsync",
+    "knative:async:run:dapr": "dapr run -H 3500 -d ../data/components/http --log-level warn",
+    "knative:async:test": "wait-on tcp:3500 tcp:8080 && curl -s -d '{\"data\": \"hello\"}' -H 'Content-Type: application/json' localhost:8080",
     "async": "concurrently npm:async:run:*",
     "async:run:func": "cross-env DEBUG=test:*,common:*,ofn:* env-cmd -e async functions-framework --target=tryAsync",
     "async:run:dapr": "dapr run -H 3500 -p 8080 -d ../data/components/cron --log-level info"

test/integration/http_binding.ts

@@ -3,12 +3,13 @@ import {deepStrictEqual} from 'assert';
 import * as sinon from 'sinon';
 import * as supertest from 'supertest';
 import * as shell from 'shelljs';
-import {Request, Response} from 'express';
 import {cloneDeep, forEach, set} from 'lodash';
 
-import {getServer} from '../../src/server';
 import {OpenFunctionContext} from '../../src/openfunction/function_context';
 
+import {OpenFunctionRuntime} from '../../src/functions';
+import {getServer} from '../../src/server';
+
 const TEST_CONTEXT: OpenFunctionContext = {
   name: 'test-context',
   version: '1.0.0',
@@ -81,15 +82,17 @@ describe('OpenFunction - HTTP Binding', () => {
       );
 
       const server = getServer(
-        (req: Request, res: Response) => {
-          res.status(200).json(TEST_PAYLOAD);
+        async (ctx: OpenFunctionRuntime, data: {}) => {
+          await ctx.send(data);
+          return ctx.response(data);
         },
         'http',
         context
       );
 
       await supertest(server)
-        .get('/')
+        .post('/')
+        .send(TEST_PAYLOAD)
         .expect(200)
         .expect(res => {
           deepStrictEqual(res.body, TEST_PAYLOAD);