Improved error catching for streams

backend/services/ChunkService/FileSystemService.ts

@@ -69,7 +69,9 @@ class FileSystemService implements ChunkInterface {
 
         const fileWriteStream = fs.createWriteStream(metadata.filePath);
 
-        await awaitUploadStreamFS(file.pipe(cipher), fileWriteStream, req, metadata.filePath);
+        const totalStreamsToErrorCatch = [file, cipher, fileWriteStream];
+
+        await awaitUploadStreamFS(file.pipe(cipher), fileWriteStream, req, metadata.filePath, totalStreamsToErrorCatch);
 
         const date = new Date();
         const encryptedFileSize = await getFileSize(metadata.filePath);
@@ -124,7 +126,9 @@ class FileSystemService implements ChunkInterface {
         res.set('Content-Disposition', 'attachment; filename="' + currentFile.filename + '"');
         res.set('Content-Length', currentFile.metadata.size.toString()); 
 
-        await awaitStream(readStream.pipe(decipher), res);
+        const allStreamsToErrorCatch = [readStream, decipher];
+
+        await awaitStream(readStream.pipe(decipher), res, allStreamsToErrorCatch);
     }
 
     getThumbnail = async(user: UserInterface, id: string) => {
@@ -148,7 +152,9 @@ class FileSystemService implements ChunkInterface {
 
         const readStream = fs.createReadStream(thumbnail.path!);
 
-        const bufferData = await streamToBuffer(readStream.pipe(decipher));
+        const allStreamsToErrorCatch = [readStream, decipher];
+
+        const bufferData = await streamToBuffer(readStream.pipe(decipher), allStreamsToErrorCatch);
 
         return bufferData;
          
@@ -177,8 +183,10 @@ class FileSystemService implements ChunkInterface {
         res.set('Content-Disposition', 'attachment; filename="' + file.filename + '"');
         res.set('Content-Length', file.metadata.size.toString());
 
+        const allStreamsToErrorCatch = [readStream, decipher];
+
         console.log("Sending Full Thumbnail...")
-        await awaitStream(readStream.pipe(decipher), res);
+        await awaitStream(readStream.pipe(decipher), res, allStreamsToErrorCatch);
         console.log("Full thumbnail sent");
     }
 
@@ -221,7 +229,9 @@ class FileSystemService implements ChunkInterface {
 
         res.writeHead(206, head);
 
-        await awaitStream(readStream.pipe(decipher), res);
+        const allStreamsToErrorCatch = [readStream, decipher];
+
+        await awaitStream(readStream.pipe(decipher), res, allStreamsToErrorCatch);
     }
 
     getPublicDownload = async(fileID: string, tempToken: any, res: Response) => {
@@ -250,7 +260,9 @@ class FileSystemService implements ChunkInterface {
         res.set('Content-Disposition', 'attachment; filename="' + file.filename + '"');
         res.set('Content-Length', file.metadata.size.toString());
 
-        await awaitStream(readStream.pipe(decipher), res);
+        const allStreamsToErrorCatch = [readStream, decipher];
+
+        await awaitStream(readStream.pipe(decipher), res, allStreamsToErrorCatch);
 
         if (file.metadata.linkType === "one") {
             console.log("removing public link");

backend/services/ChunkService/MongoService.ts

@@ -71,7 +71,9 @@ class MongoService implements ChunkInterface {
                 
         bucketStream = bucket.openUploadStream(filename, {metadata});
 
-        const finishedFile = await awaitUploadStream(file.pipe(cipher), bucketStream, req) as FileInterface;
+        const allStreamsToErrorCatch = [file, cipher, bucketStream];
+
+        const finishedFile = await awaitUploadStream(file.pipe(cipher), bucketStream, req, allStreamsToErrorCatch) as FileInterface;
 
         const imageCheck = imageChecker(filename);
  
@@ -110,7 +112,9 @@ class MongoService implements ChunkInterface {
         res.set('Content-Disposition', 'attachment; filename="' + currentFile.filename + '"');
         res.set('Content-Length', currentFile.metadata.size.toString()); 
 
-        await awaitStream(readStream.pipe(decipher), res);
+        const allStreamsToErrorCatch = [readStream, decipher];
+
+        await awaitStream(readStream.pipe(decipher), res, allStreamsToErrorCatch);
     }
 
     getThumbnail = async(user: UserInterface, id: string) => {
@@ -164,7 +168,10 @@ class MongoService implements ChunkInterface {
         res.set('Content-Length', file.metadata.size.toString());
 
         console.log("Sending Full Thumbnail...")
-        await awaitStream(readStream.pipe(decipher), res);
+
+        const allStreamsToErrorCatch = [readStream, decipher];
+
+        await awaitStream(readStream.pipe(decipher), res, allStreamsToErrorCatch);
         console.log("Full thumbnail sent");
     }
 
@@ -196,7 +203,9 @@ class MongoService implements ChunkInterface {
         res.set('Content-Disposition', 'attachment; filename="' + file.filename + '"');
         res.set('Content-Length', file.metadata.size.toString());
 
-        await awaitStream(readStream.pipe(decipher), res);
+        const allStreamsToErrorCatch = [readStream, decipher];
+
+        await awaitStream(readStream.pipe(decipher), res, allStreamsToErrorCatch);
 
         if (file.metadata.linkType === "one") {
             console.log("removing public link");
@@ -247,7 +256,9 @@ class MongoService implements ChunkInterface {
 
         res.writeHead(206, head);
 
-        await awaitStream(readStream.pipe(decipher), res);
+        const allStreamsToErrorCatch = [readStream, decipher];
+
+        await awaitStream(readStream.pipe(decipher), res, allStreamsToErrorCatch);
     }
 
     deleteFile = async(userID: string, fileID: string) => {

backend/services/ChunkService/S3Service.ts

@@ -125,7 +125,9 @@ class S3Service implements ChunkInterface {
 
         const s3ReadStream = s3.getObject(params).createReadStream();
 
-        await awaitStream(s3ReadStream.pipe(decipher), res);
+        const allStreamsToErrorCatch = [s3ReadStream, decipher];
+
+        await awaitStream(s3ReadStream.pipe(decipher), res, allStreamsToErrorCatch);
 
     }
 
@@ -167,7 +169,9 @@ class S3Service implements ChunkInterface {
 
         res.writeHead(206, head);
 
-        await awaitStream(s3ReadStream.pipe(decipher), res);
+        const allStreamsToErrorCatch = [s3ReadStream, decipher];
+
+        await awaitStream(s3ReadStream.pipe(decipher), res, allStreamsToErrorCatch);
         
     }
 
@@ -194,7 +198,9 @@ class S3Service implements ChunkInterface {
 
         const readStream = s3.getObject(params).createReadStream();
 
-        const bufferData = await streamToBuffer(readStream.pipe(decipher));
+        const allStreamsToErrorCatch = [readStream, decipher];
+
+        const bufferData = await streamToBuffer(readStream.pipe(decipher), allStreamsToErrorCatch);
 
         return bufferData;
     } 
@@ -224,8 +230,10 @@ class S3Service implements ChunkInterface {
         res.set('Content-Disposition', 'attachment; filename="' + file.filename + '"');
         res.set('Content-Length', file.metadata.size.toString());
 
+        const allStreamsToErrorCatch = [readStream, decipher];
+
         console.log("Sending Full Thumbnail...")
-        await awaitStream(readStream.pipe(decipher), res);
+        await awaitStream(readStream.pipe(decipher), res, allStreamsToErrorCatch);
         console.log("Full thumbnail sent");
     }
 
@@ -257,7 +265,9 @@ class S3Service implements ChunkInterface {
         res.set('Content-Disposition', 'attachment; filename="' + file.filename + '"');
         res.set('Content-Length', file.metadata.size.toString());
 
-        await awaitStream(readStream.pipe(decipher), res);
+        const allStreamsToErrorCatch = [readStream, decipher];
+
+        await awaitStream(readStream.pipe(decipher), res, allStreamsToErrorCatch);
 
         if (file.metadata.linkType === "one") {
             console.log("removing public link");

backend/services/ChunkService/utils/awaitStream.ts

@@ -1,21 +1,17 @@
-const awaitStream = <T>(inputSteam: any, outputStream: any) => {
+const awaitStream = <T>(inputSteam: any, outputStream: any, allStreamsToErrorCatch: any[]) => {
 
     return new Promise<T>((resolve, reject) => {
 
-        inputSteam.on("error", (e: Error) => {
-            reject({
-                message: "Await Stream Input Error",
-                code: 500,
-                error: e
-            })
-        })
+        allStreamsToErrorCatch.forEach((currentStream: any) => {
 
-        outputStream.on("error", (e: Error) => {
-            reject({
-                message: "Await Stream Output Error",
-                code: 500,
-                error: e
+            currentStream.on("error", (e: Error) => {
+                reject({
+                    message: "Await Stream Input Error",
+                    code: 500,
+                    error: e
+                })
             })
+            
         })
 
         inputSteam.pipe(outputStream).on("finish", (data: T) => {

backend/services/ChunkService/utils/awaitUploadStream.ts

@@ -1,31 +1,45 @@
 import removeChunks from "../../FileService/utils/removeChunks";
 import {Request} from "express"
 
-const awaitUploadStream = <T>(inputSteam: any, outputStream: any, req: Request) => {
+const awaitUploadStream = <T>(inputSteam: any, outputStream: any, req: Request, allStreamsToErrorCatch: any[]) => {
 
     return new Promise<T>((resolve, reject) => {
 
-        inputSteam.on("error", (e: Error) => {
+        allStreamsToErrorCatch.forEach((currentStream: any) => {
 
-            removeChunks(outputStream);
+            currentStream.on("error", (e: Error) => {
+
+                removeChunks(outputStream);
+        
+                reject({
+                    message: "Await Stream Input Error",
+                    code: 500,
+                    error: e
+                })
+            })
+        })
+
+        // inputSteam.on("error", (e: Error) => {
+
+        //     removeChunks(outputStream);
             
-            reject({
-                message: "Await Stream Input Error",
-                code: 500,
-                error: e
-            })
-        })
+        //     reject({
+        //         message: "Await Stream Input Error",
+        //         code: 500,
+        //         error: e
+        //     })
+        // })
 
-        outputStream.on("error", (e: Error) => {
+        // outputStream.on("error", (e: Error) => {
 
-            removeChunks(outputStream);
+        //     removeChunks(outputStream);
 
-            reject({
-                message: "Await Stream Output Error",
-                code: 500,
-                error: e
-            })
-        })
+        //     reject({
+        //         message: "Await Stream Output Error",
+        //         code: 500,
+        //         error: e
+        //     })
+        // })
 
         req.on("aborted", () => {
 

backend/services/ChunkService/utils/awaitUploadStreamFS.ts

@@ -9,6 +9,7 @@ export const removeChunksFS = (path: string) => {
 
             if (err) {
                 console.log("Could not remove fs file", err);
+                resolve();
             }
 
             console.log("File Removed");
@@ -17,31 +18,45 @@ export const removeChunksFS = (path: string) => {
     })
 }
 
-const awaitUploadStream = <T>(inputSteam: any, outputStream: any, req: Request, path: string) => {
+const awaitUploadStream = <T>(inputSteam: any, outputStream: any, req: Request, path: string, allStreamsToCatchError: any[]) => {
 
     return new Promise<T>((resolve, reject) => {
+        
+        allStreamsToCatchError.forEach((currentStream: any) => {
 
-        inputSteam.on("error", (e: Error) => {
+            currentStream.on("error", (e: Error) => {
 
-            removeChunksFS(path);
+                removeChunksFS(path);
             
-            reject({
-                message: "Await Stream Input Error",
-                code: 500,
-                error: e
+                reject({
+                    message: "Await Stream Input Error",
+                    code: 500,
+                    error: e
+                })
             })
         })
 
-        outputStream.on("error", (e: Error) => {
+        // inputSteam.on("error", (e: Error) => {
 
-            removeChunksFS(path);
+        //     removeChunksFS(path);
+            
+        //     reject({
+        //         message: "Await Stream Input Error",
+        //         code: 500,
+        //         error: e
+        //     })
+        // })
 
-            reject({
-                message: "Await Stream Output Error",
-                code: 500,
-                error: e
-            })
-        })
+        // outputStream.on("error", (e: Error) => {
+
+        //     removeChunksFS(path);
+
+        //     reject({
+        //         message: "Await Stream Output Error",
+        //         code: 500,
+        //         error: e
+        //     })
+        // })
 
         req.on("aborted", () => {
 

backend/utils/streamToBuffer.ts

@@ -1,10 +1,27 @@
-const streamToBuffer = (stream: any) => {
-    const chunks: any[] = []
-    return new Promise<Buffer>((resolve, reject) => {
-      stream.on('data', (chunk: any) => chunks.push(chunk))
-      stream.on('error', reject)
-      stream.on('end', () => resolve(Buffer.concat(chunks)))
+const streamToBuffer = (stream: any, allStreamsToErrorCatch: any[]) => {
+
+  const chunks: any[] = []
+  return new Promise<Buffer>((resolve, reject) => {
+
+    allStreamsToErrorCatch.forEach((currentStream) => {
+
+      currentStream.on("error", (e: Error) => {
+  
+        console.log("Stream To Buffer Error", e);
+        reject({
+          message: "stream to buffer error",
+          code: 500,
+          error: e
+         })
+  
+      })
     })
+
+  stream.on('data', (chunk: any) => chunks.push(chunk));
+  stream.on('error', reject);
+  stream.on('end', () => resolve(Buffer.concat(chunks)));
+
+  })
 }
 
 export default streamToBuffer;