Merge pull request #1024 from mozilla/ws

implemented websocket flow control to prevent over buffering

package.json

@@ -140,7 +140,7 @@
     "cldr-core": "^34.0.0",
     "convict": "^4.4.0",
     "express": "^4.16.3",
-    "express-ws": "^4.0.0",
+    "express-ws": "dannycoates/express-ws",
     "fluent": "^0.9.1",
     "fluent-langneg": "^0.1.0",
     "helmet": "^3.15.0",

server/routes/ws.js

@@ -3,10 +3,11 @@ const storage = require('../storage');
 const config = require('../config');
 const mozlog = require('../log');
 const Limiter = require('../limiter');
-const Parser = require('../streamparser');
 const wsStream = require('websocket-stream/stream');
 const fxa = require('../fxa');
 
+const { Duplex } = require('stream');
+
 const log = mozlog('send.upload');
 
 module.exports = function(ws, req) {
@@ -72,12 +73,27 @@ module.exports = function(ws, req) {
           id: newId
         })
       );
-
       const limiter = new Limiter(maxFileSize);
-      const parser = new Parser();
+      const flowControl = new Duplex({
+        read() {
+          ws.resume();
+        },
+        write(chunk, encoding, callback) {
+          if (chunk.length === 1 && chunk[0] === 0) {
+            this.push(null);
+          } else {
+            if (!this.push(chunk)) {
+              ws.pause();
+            }
+          }
+          callback();
+        }
+      });
+
       fileStream = wsStream(ws, { binary: true })
         .pipe(limiter)
-        .pipe(parser);
+        .pipe(flowControl);
+
       await storage.set(newId, fileStream, meta, timeLimit);
 
       if (ws.readyState === 1) {

server/streamparser.js

@@ -1,15 +0,0 @@
-const { Duplex } = require('stream');
-
-class StreamParser extends Duplex {
-  _write(chunk, encoding, callback) {
-    if (chunk.byteLength === 1 && chunk[0] === 0) {
-      this.push(null);
-    } else {
-      this.push(chunk);
-    }
-    callback();
-  }
-  _read() {}
-}
-
-module.exports = StreamParser;