use a Duplex stream for EOF

app/fileManager.js

@@ -52,7 +52,7 @@ export default function(state, emitter) {
     checkFiles();
   });
 
-  //emitter.on('navigate', checkFiles);
+  emitter.on('navigate', checkFiles);
 
   emitter.on('render', () => {
     lastRender = Date.now();

server/routes/ws.js

@@ -50,26 +50,35 @@ module.exports = async function(ws, req) {
       fileStream = wsStream(ws, { binary: true })
         .pipe(limiter)
         .pipe(parser);
-      storage.set(newId, fileStream, meta);
+      await storage.set(newId, fileStream, meta);
 
-      await parser.promise;
+      if (ws.readyState === 1) {
+        // if the socket is closed by a cancelled upload the stream
+        // ends without an error so we need to check the state
+        // before sending a reply.
 
-      ws.send(
-        JSON.stringify({
-          url,
-          owner: meta.owner,
-          id: newId,
-          authentication: `send-v1 ${meta.nonce}`
-        })
-      );
+        // TODO: we should handle cancelled uploads differently
+        // in order to avoid having to check socket state and clean
+        // up storage, possibly with an exception that we can catch.
+        ws.send(
+          JSON.stringify({
+            url,
+            owner: meta.owner,
+            id: newId,
+            authentication: `send-v1 ${meta.nonce}`
+          })
+        );
+      }
     } catch (e) {
       log.error('upload', e);
-      ws.send(
-        JSON.stringify({
-          error: e === 'limit' ? 413 : 500
-        })
-      );
-      ws.close();
+      if (ws.readyState === 1) {
+        ws.send(
+          JSON.stringify({
+            error: e === 'limit' ? 413 : 500
+          })
+        );
+        ws.close();
+      }
     }
   });
 };

server/streamparser.js

@@ -1,23 +1,15 @@
-const { Transform } = require('stream');
+const { Duplex } = require('stream');
 
-class StreamParser extends Transform {
-  constructor() {
-    super();
-    let res;
-    this.promise = new Promise(resolve => {
-      res = resolve;
-    });
-    this.res = res;
-  }
-
-  _transform(chunk, encoding, callback) {
+class StreamParser extends Duplex {
+  _write(chunk, encoding, callback) {
     if (chunk.byteLength === 1 && chunk[0] === 0) {
-      this.res();
+      this.push(null);
     } else {
       this.push(chunk);
     }
     callback();
   }
+  _read() {}
 }
 
 module.exports = StreamParser;