support streaming

Author: vdtn359

.npmrc

@@ -0,0 +1,2 @@
+@stackla:registry=https://npm.pkg.github.com
+//npm.pkg.github.com/:_authToken=${NODE_AUTH_TOKEN}

lib/framework/get-framework.js

@@ -3,62 +3,62 @@
 const http = require('http')
 const Response = require('../response');
 
-function common(cb) {
+function common(cb, opts) {
   return request => {
-    const response = new Response(request);
+    const response = new Response(request, opts);
 
     cb(request, response);
 
     return response;
   };
 }
 
-module.exports = function getFramework(app) {
+module.exports = function getFramework(app, opts) {
   if (app instanceof http.Server) {
     return request => {
-      const response = new Response(request);
+      const response = new Response(request, opts);
       app.emit('request', request, response)
       return response
     }
   }
 
   if (typeof app.callback === 'function') {
-    return common(app.callback());
+    return common(app.callback(), opts);
   }
 
   if (typeof app.handle === 'function') {
     return common((request, response) => {
       app.handle(request, response);
-    });
+    }, opts);
   }
 
   if (typeof app.handler === 'function') {
     return common((request, response) => {
       app.handler(request, response);
-    });
+    }, opts);
   }
 
   if (typeof app._onRequest === 'function') {
     return common((request, response) => {
       app._onRequest(request, response);
-    });
+    }, opts);
   }
 
   if (typeof app === 'function') {
-    return common(app);
+    return common(app, opts);
   }
 
   if (app.router && typeof app.router.route == 'function') {
     return common((req, res) => {
       const { url, method, headers, body } = req;
       app.router.route({ url, method, headers, body }, res);
-    });
+    }, opts);
   }
 
   if (app._core && typeof app._core._dispatch === 'function') {
     return common(app._core._dispatch({
       app
-    }));
+    }), opts);
   }
 
   if (typeof app.inject === 'function') {
@@ -72,7 +72,7 @@ module.exports = function getFramework(app) {
   }
 
   if (typeof app.main === 'function') {
-    return common(app.main);
+    return common(app.main, opts);
   }
 
   throw new Error('Unsupported framework');

lib/provider/aws/format-response.js

@@ -26,22 +26,26 @@ module.exports = (event, response, options) => {
   if (multiValueHeaders['set-cookie']) {
     cookies = multiValueHeaders['set-cookie'];
   }
+  const isBase64Encoded = !response.streaming && isBinary(headers, options);
+  let body;
+  if (response.streaming) {
+    body = Response.stream(response);
+  } else {
+    const encoding = isBase64Encoded ? 'base64' : 'utf8';
+    body = Response.body(response).toString(encoding);
 
-  const isBase64Encoded = isBinary(headers, options);
-  const encoding = isBase64Encoded ? 'base64' : 'utf8';
-  let body = Response.body(response).toString(encoding);
-
-  if (headers['transfer-encoding'] === 'chunked' || response.chunkedEncoding) {
-    const raw = Response.body(response).toString().split('\r\n');
-    const parsed = [];
-    for (let i = 0; i < raw.length; i +=2) {
-      const size = parseInt(raw[i], 16);
-      const value = raw[i + 1];
-      if (value) {
-        parsed.push(value.substring(0, size));
+    if (headers['transfer-encoding'] === 'chunked' || response.chunkedEncoding) {
+      const raw = Response.body(response).toString().split('\r\n');
+      const parsed = [];
+      for (let i = 0; i < raw.length; i +=2) {
+        const size = parseInt(raw[i], 16);
+        const value = raw[i + 1];
+        if (value) {
+          parsed.push(value.substring(0, size));
+        }
       }
+      body = parsed.join('')
     }
-    body = parsed.join('')
   }
 
   if (eventType === LAMBDA_EVENT_TYPES.ALB) {

lib/response.js

@@ -1,11 +1,13 @@
 'use strict';
 
 const http = require('http');
+const stream = require('stream');
 
 const headerEnd = '\r\n\r\n';
 
 const BODY = Symbol();
 const HEADERS = Symbol();
+const STREAM = Symbol();
 
 function getString(data) {
   if (Buffer.isBuffer(data)) {
@@ -19,7 +21,11 @@ function getString(data) {
 
 function addData(stream, data) {
   if (Buffer.isBuffer(data) || typeof data === 'string' || data instanceof Uint8Array) {
-    stream[BODY].push(Buffer.from(data));
+    if (stream.streaming) {
+      stream[STREAM].push(Buffer.from(data));
+    } else {
+      stream[BODY].push(Buffer.from(data));
+    }
   } else {
     throw new Error(`response.write() of unexpected type: ${typeof data}`);
   }
@@ -42,6 +48,10 @@ module.exports = class ServerlessResponse extends http.ServerResponse {
     return Buffer.concat(res[BODY]);
   }
 
+  static stream(res) {
+    return res[STREAM];
+  }
+
   static headers(res) {
     const headers = typeof res.getHeaders === 'function'
       ? res.getHeaders()
@@ -80,11 +90,15 @@ module.exports = class ServerlessResponse extends http.ServerResponse {
     super.writeHead(statusCode, reason, obj);
   }
 
-  constructor({ method }) {
+  constructor({ method }, { streaming = false } = {}) {
     super({ method });
 
     this[BODY] = [];
     this[HEADERS] = {};
+    if (streaming) {
+      this[STREAM] = stream.PassThrough();
+      this.streaming = streaming;
+    }
 
     this.useChunkedEncodingByDefault = false;
     this.chunkedEncoding = false;

package-lock.json

@@ -1,6 +1,6 @@
 {
-  "name": "serverless-http",
-  "version": "3.2.0",
+  "name": "@stackla/serverless-http",
+  "version": "3.2.1",
   "description": "Use existing web application frameworks in serverless environments",
   "main": "serverless-http.js",
   "types": "serverless-http.d.ts",
@@ -19,8 +19,7 @@
     "test": "nyc mocha",
     "posttest": "eslint lib test",
     "test:integration:offline": "mocha test/integration/test-offline.js --exit",
-    "test:integration:aws": "mocha test/integration/test-aws.js --exit",
-    "postpublish": "git push origin master --tags"
+    "test:integration:aws": "mocha test/integration/test-aws.js --exit"
   },
   "keywords": [
     "serverless",
@@ -93,5 +92,8 @@
     "supertest": "^6.2.2",
     "tree-kill": "^1.2.2",
     "typescript": "^4.6.3"
+  },
+  "dependencies": {
+    "is-stream": "^3.0.0"
   }
 }

package.json

@@ -24,6 +24,7 @@ declare namespace ServerlessHttp {
     request?: Object | Function,
     response?: Object | Function,
     binary?: boolean | Function | string | string[],
+    streaming?: boolean,
     basePath?: string
   }
   /**

serverless-http.d.ts

@@ -5,13 +5,14 @@ const getFramework = require('./lib/framework/get-framework');
 const getProvider = require('./lib/provider/get-provider');
 
 const defaultOptions = {
-  requestId: 'x-request-id'
+  requestId: 'x-request-id',
+  streaming: false,
 };
 
 module.exports = function (app, opts) {
   const options = Object.assign({}, defaultOptions, opts);
 
-  const framework = getFramework(app);
+  const framework = getFramework(app, opts);
   const provider = getProvider(options);
 
   return provider(async (request, ...context) => {