implemented websocket flow control to prevent over buffering

This commit is contained in:
Danny Coates 2018-11-22 13:21:39 -08:00
parent 015067648e
commit 531584dbf1
No known key found for this signature in database
GPG Key ID: 4C442633C62E00CB
4 changed files with 207 additions and 191 deletions

357
package-lock.json generated

File diff suppressed because it is too large Load Diff

View File

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

View File

@ -3,10 +3,11 @@ const storage = require('../storage');
const config = require('../config'); const config = require('../config');
const mozlog = require('../log'); const mozlog = require('../log');
const Limiter = require('../limiter'); const Limiter = require('../limiter');
const Parser = require('../streamparser');
const wsStream = require('websocket-stream/stream'); const wsStream = require('websocket-stream/stream');
const fxa = require('../fxa'); const fxa = require('../fxa');
const { Duplex } = require('stream');
const log = mozlog('send.upload'); const log = mozlog('send.upload');
module.exports = function(ws, req) { module.exports = function(ws, req) {
@ -72,12 +73,27 @@ module.exports = function(ws, req) {
id: newId id: newId
}) })
); );
const limiter = new Limiter(maxFileSize); 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 }) fileStream = wsStream(ws, { binary: true })
.pipe(limiter) .pipe(limiter)
.pipe(parser); .pipe(flowControl);
await storage.set(newId, fileStream, meta, timeLimit); await storage.set(newId, fileStream, meta, timeLimit);
if (ws.readyState === 1) { if (ws.readyState === 1) {

View File

@ -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;