Merge pull request #122 from ente-io/fix-file-stream-stuck

fix zip file stream getting stuck

src/services/fs.ts

@@ -4,6 +4,7 @@ import * as fs from 'promise-fs';
 import { ElectronFile } from '../types';
 import StreamZip from 'node-stream-zip';
 import { Readable } from 'stream';
+import { logError } from './logging';
 
 // https://stackoverflow.com/a/63111390
 export const getDirFilePaths = async (dirPath: string) => {
@@ -97,31 +98,50 @@ export const getZipFileStream = async (
     const done = {
         current: false,
     };
+    const inProgress = {
+        current: false,
+    };
     let resolveObj: (value?: any) => void = null;
     let rejectObj: (reason?: any) => void = null;
     stream.on('readable', () => {
-        if (resolveObj) {
-            const chunk = stream.read(FILE_STREAM_CHUNK_SIZE) as Buffer;
-
-            if (chunk) {
-                resolveObj(new Uint8Array(chunk));
-                resolveObj = null;
+        try {
+            if (resolveObj) {
+                inProgress.current = true;
+                const chunk = stream.read(FILE_STREAM_CHUNK_SIZE) as Buffer;
+                if (chunk) {
+                    resolveObj(new Uint8Array(chunk));
+                    resolveObj = null;
+                }
+                inProgress.current = false;
             }
+        } catch (e) {
+            rejectObj(e);
         }
     });
     stream.on('end', () => {
-        done.current = true;
+        try {
+            done.current = true;
+            if (resolveObj && !inProgress.current) {
+                resolveObj(null);
+                resolveObj = null;
+            }
+        } catch (e) {
+            rejectObj(e);
+        }
     });
     stream.on('error', (e) => {
-        done.current = true;
-
-        if (rejectObj) {
+        try {
+            done.current = true;
+            if (rejectObj) {
+                rejectObj(e);
+                rejectObj = null;
+            }
+        } catch (e) {
             rejectObj(e);
-            rejectObj = null;
         }
     });
 
-    const readStreamData = () => {
+    const readStreamData = async () => {
         return new Promise<Uint8Array>((resolve, reject) => {
             const chunk = stream.read(FILE_STREAM_CHUNK_SIZE) as Buffer;
 
@@ -145,6 +165,7 @@ export const getZipFileStream = async (
                     controller.close();
                 }
             } catch (e) {
+                logError(e, 'readableStream pull failed');
                 controller.close();
             }
         },

ui

@@ -1 +1 @@
-Subproject commit 1f0caeedb8e3b5c9027282eec3d70c6021a24556
+Subproject commit d54df9d068e11f75d7ea4dfb0ea3ef768b3cc64e