feat: Streaming connections can now be defined in the accept.json file, adding tests, headers are now transferred, proper error handling, started adding text to the readme

mladkau · mladkau · commit 3c3b24ecebec · 2019-10-18T11:20:28.000+01:00
diff --git a/README.npm.md b/README.npm.md
@@ -117,6 +117,22 @@ The `private` rules should be determined by what you want to allow through the s
 
 This `public` rule will ensure everything is forwarded to your clients, and will allow your client to handle blocking out messages.
 
+Sometimes large amounts of data need to be transferred in a response to a request. In these instances it might makes sense to stream
+the data for these requests:
+
+```json
+  "public": [{
+    "//": "accept any requests to our connected clients",
+    "method": "any",
+    "path": "/blobs/*",
+    "stream": true
+  }, {
+    "//": "accept any requests to our connected clients",
+    "method": "any",
+    "path": "/*"
+  }],
+```
+
 ## Development & how to test
 
 The project's source code is written in full ES6 (with commonjs modules). This requires the source to be developed with node@6. However, during the release process, the code is transpiled to ES5 via babel and is released with node LTS in mind, node@4 and upwards.
@@ -284,4 +300,3 @@ Public filters are for requests that a recieved on your broker client and are in
 * [License: Apache License, Version 2.0](LICENSE)
 * [Contributing](.github/CONTRIBUTING.md)
 * [Security](SECURITY.md)
-
diff --git a/lib/filters/index.js b/lib/filters/index.js
@@ -32,7 +32,7 @@ module.exports = ruleSource => {
   // array of entries with
   const tests = rules.map(entry => {
     const keys = [];
-    let { method, origin, path, valid } = entry;
+    let { method, origin, path, valid, stream } = entry;
     method = (method || 'get').toLowerCase();
     valid = valid || [];
 
@@ -135,6 +135,7 @@ module.exports = ruleSource => {
       return {
         url: origin + url + querystring,
         auth: entry.auth && authHeader(entry.auth),
+        stream
       };
     };
   });
diff --git a/lib/relay.js b/lib/relay.js
@@ -9,36 +9,44 @@ const tryJSONParse = require('./try-json-parse');
 const logger = require('./log');
 const version = require('./version');
 const stream = require('stream');
+const NodeCache = require('node-cache');
 
 module.exports = {
   request: requestHandler,
   response: responseHandler,
   streamingResponse: streamResponseHandler,
 };
 
-const binaryRequest = [
-  'application/vnd.docker.image.rootfs.diff.tar.gzip',
-];
+const streams = new NodeCache({
+  stdTTL: 18000,  // 5 hours
+  checkperiod: 3600, // 1 hour
+  useClones: false
+});
 
-const streams = new Map();
-
-// Stream processing is only available on
-// the server side i.e. the client sends
-// binary chunks to the server.
 function streamResponseHandler(token) {
-  return (streamingID, chunk, finished) => {
-    var streamBuffer = streams.get(streamingID);
+  return (streamingID, chunk, finished, ioResponse) => {
+    const stream = streams.get(streamingID);
 
-    if (streamBuffer) {
-      if (chunk) {
-        streamBuffer.write(chunk);
-      }
-      if (finished) {
-        streamBuffer.end();
-        streams.delete(streamingID);
+    if (stream) {
+      const { streamBuffer, response } = stream;
+
+      if (streamBuffer) {
+        if (ioResponse) {
+          response.status(ioResponse.status)
+            .set(ioResponse.headers);
+        }
+        if (chunk) {
+          streamBuffer.write(chunk);
+        }
+        if (finished) {
+          streamBuffer.end();
+          streams.del(streamingID);
+        }
+      } else {
+        logger.warn({ streamingID, token }, 'discarding binary chunk');
       }
     } else {
-      logger.warn({ streamingID, token }, 'discarding binary chunk');
+      logger.warn({ streamingID, token }, 'trying to write into a closed stream');
     }
   };
 }
@@ -74,14 +82,16 @@ function requestHandler(filterRules) {
       logContext.ioUrl = result.url;
 
       // check if this is a streaming request for binary data
-      if (req.headers && binaryRequest.indexOf(req.headers['accept']) !== -1) {
+      if (result.stream) {
         const streamingID = uuid();
         const streamBuffer = new stream.PassThrough();
-
         logContext.streamingID = streamingID;
-        logger.debug(logContext, 'sending binary request over websocket connection');
+        logger.debug(logContext, 'sending stream request over websocket connection');
 
-        streams.set(streamingID, streamBuffer);
+        streams.set(streamingID, {
+          response : res,
+          streamBuffer
+        });
         streamBuffer.pipe(res);
 
         res.locals.io.send('request', {
@@ -90,6 +100,12 @@ function requestHandler(filterRules) {
           body: req.body,
           headers: req.headers,
           streamingID,
+        }, () => {
+          // Streaming requests should not be handled using the emit function
+          // but rather by sending 'chunk' messages
+          const msg = 'Broker client does not support streaming requests';
+          logger.error(logContext, msg);
+          return res.status(501).send({message: msg});
         });
 
         return;
@@ -263,41 +279,60 @@ function responseHandler(filterRules, config, io) {
 
       // check if this is a streaming request for binary data
       if (streamingID) {
-        logger.debug(logContext, 'serving binary stream request');
+        logger.debug(logContext, 'serving stream request');
 
         req.encoding = null; // no encoding for binary data
 
-        request(req).on('data', chunk => {
+        request(req).on('response', response => {
+          const status = (response && response.statusCode) || 500;
+          logResponse(logContext, status, response.headers);
+          io.send('chunk', streamingID, '', false, {
+            status,
+            headers : response.headers
+          });
+        }).on('data', chunk => {
           io.send('chunk', streamingID, chunk, false);
         }).on('end', () => {
           io.send('chunk', streamingID, '', true);
+        }).on('error', error => {
+          logError(logContext, error);
+          io.send('chunk', streamingID, error.message, true, {
+            status : 500
+          });
         });
         return;
       }
 
       request(req, (error, response, body) => {
         if (error) {
-          logContext.error = error;
-          logger.error(logContext, 'error while sending websocket request over HTTP connection');
+          logError(logContext, error);
           return emit({
             status: 500,
             body: error.message
           });
         }
 
         const status = (response && response.statusCode) || 500;
-        logContext.httpStatus = status;
-        logContext.httpHeaders = response.headers;
-
-        const logMsg = 'sending response back to websocket connection';
-        if (status <= 200) {
-          logger.debug(logContext, logMsg);
-        } else {
-          logger.info(logContext, logMsg);
-        }
-
+        logResponse(logContext, status, response.headers);
         emit({ status, body, headers: response.headers });
       });
     });
   };
 }
+
+function logResponse(logContext, status, headers) {
+  logContext.httpStatus = status;
+  logContext.httpHeaders = headers;
+
+  const logMsg = 'sending response back to websocket connection';
+  if (status <= 200) {
+    logger.debug(logContext, logMsg);
+  } else {
+    logger.info(logContext, logMsg);
+  }
+}
+
+function logError(logContext, error) {
+  logContext.error = error;
+  logger.error(logContext, 'error while sending websocket request over HTTP connection');
+}
diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -46,6 +46,7 @@
     "lodash.escaperegexp": "^4.1.2",
     "lodash.mapvalues": "^4.6.0",
     "minimatch": "^3.0.4",
+    "node-cache": "^4.2.1",
     "path-to-regexp": "^1.5.3",
     "primus": "^6.0.1",
     "primus-emitter": "^3.1.1",
diff --git a/test/fixtures/client/filters.json b/test/fixtures/client/filters.json
@@ -72,6 +72,11 @@
       ]
     },
 
+    {
+      "path": "/test-blob/*",
+      "method": "GET",
+      "origin": "http://localhost:${originPort}"
+    },
 
     {
       "path": "/repos/:repo/:owner/contents/folder*/package.json",
diff --git a/test/fixtures/server/filters.json b/test/fixtures/server/filters.json
@@ -28,10 +28,21 @@
       "path": "/echo-query/:param?",
       "method": "GET",
       "origin": "http://localhost:${originPort}"
+    },
+
+    {
+      "path": "/test-blob/*",
+      "method": "GET",
+      "origin": "http://localhost:${originPort}",
     }
 
   ],
   "public": [
+    {
+      "path": "/test-blob/*",
+      "method": "GET",
+      "stream": true
+    },
     {
       "path": "/*",
       "method": "GET"
diff --git a/test/functional/server-client.test.js b/test/functional/server-client.test.js
@@ -44,7 +44,7 @@ test('proxy requests originating from behind the broker server', t => {
   server.io.on('connection', socket => {
     socket.on('identify', clientData => {
       const token = clientData.token;
-      t.plan(22);
+      t.plan(24);
 
       t.test('identification', t => {
         const filters = require(`${clientRootPath}/${ACCEPT}`);
@@ -288,6 +288,49 @@ test('proxy requests originating from behind the broker server', t => {
         });
       });
 
+      t.test('successfully stream data', t => {
+        const url = `http://localhost:${serverPort}/broker/${token}/test-blob/1`;
+        request({
+          url,
+          method: 'get',
+          encoding: null
+        }, (err, res, body) => {
+
+          // No encoding is only possible when streaming
+          // data as we otherwise encode the data
+          // when making the request on the client.
+
+          t.equal(res.statusCode, 299, '299 statusCode');
+          t.equal(res.headers['test-orig-url'],
+            '/test-blob/1', 'orig URL');
+
+          // Check that the server response with the correct data
+
+          const buf = new Buffer(500);
+          for (var i=0; i<500; i++) {
+            buf.writeUInt8(i & 0xFF, i);
+          }
+          t.deepEqual(body, buf);
+
+          t.end();
+        });
+      });
+
+      t.test('fail to stream data', t => {
+        const url = `http://localhost:${serverPort}/broker/${token}/test-blob/2`;
+        request({
+          url,
+          method: 'get',
+          encoding: null
+        }, (err, res, body) => {
+          t.equal(res.statusCode, 500, '500 statusCode');
+          t.equal(res.headers['test-orig-url'],
+            '/test-blob/2', 'orig URL');
+          t.equal(String(body), 'Test Error');
+          t.end();
+        });
+      });
+
       t.test('clean up', t => {
         client.close();
         setTimeout(() => {
diff --git a/test/utils.js b/test/utils.js
@@ -15,6 +15,27 @@ const { app: echoServer, server } = webserver({
   httpsCert: process.env.TEST_CERT, // Optional
 });
 
+echoServer.get(
+  '/test-blob/1',
+  (req, res) => {
+    res.setHeader('test-orig-url', req.originalUrl);
+    res.status(299);
+
+    const buf = new Buffer(500);
+    for (var i=0; i<500; i++) {
+      buf.writeUInt8(i & 0xFF, i);
+    }
+    res.send(buf);
+  });
+
+echoServer.get(
+  '/test-blob/2',
+  (req, res) => {
+    res.setHeader('test-orig-url', req.originalUrl);
+    res.status(500);
+    res.send('Test Error');
+  });
+
 echoServer.get('/basic-auth', (req, res) => {
   res.send(req.headers.authorization);
 });
@@ -44,7 +65,7 @@ echoServer.get(
   (req, res) => {
     res.send(req.originalUrl);
   });
-  
+
 echoServer.get('/repos/owner/repo/contents/folder/package.json',
   (req, res) => {
     res.json({headers: req.headers, query: req.query, url: req.url});

Original file line number	Diff line number	Diff line change
`@@ -72,6 +72,11 @@`
`72`	`72`	`]`
`73`	`73`	`},`
`74`	`74`
	`75`	`+ {`
	`76`	`+ "path": "/test-blob/*",`
	`77`	`+ "method": "GET",`
	`78`	`+ "origin": "http://localhost:${originPort}"`
	`79`	`+ },`
`75`	`80`
`76`	`81`	`{`
`77`	`82`	`"path": "/repos/:repo/:owner/contents/folder*/package.json",`