diff --git a/backend/services/ChunkService/FileSystemService.ts b/backend/services/ChunkService/FileSystemService.ts index c7a570d..d4d408b 100644 --- a/backend/services/ChunkService/FileSystemService.ts +++ b/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"); diff --git a/backend/services/ChunkService/MongoService.ts b/backend/services/ChunkService/MongoService.ts index 16a39f3..07a0496 100644 --- a/backend/services/ChunkService/MongoService.ts +++ b/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) => { diff --git a/backend/services/ChunkService/S3Service.ts b/backend/services/ChunkService/S3Service.ts index 2b8c479..f3f5606 100644 --- a/backend/services/ChunkService/S3Service.ts +++ b/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"); diff --git a/backend/services/ChunkService/utils/awaitStream.ts b/backend/services/ChunkService/utils/awaitStream.ts index 682b671..1b01c8d 100644 --- a/backend/services/ChunkService/utils/awaitStream.ts +++ b/backend/services/ChunkService/utils/awaitStream.ts @@ -1,21 +1,17 @@ -const awaitStream = (inputSteam: any, outputStream: any) => { +const awaitStream = (inputSteam: any, outputStream: any, allStreamsToErrorCatch: any[]) => { return new Promise((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) => { diff --git a/backend/services/ChunkService/utils/awaitUploadStream.ts b/backend/services/ChunkService/utils/awaitUploadStream.ts index 5a7f0d3..9526ad7 100644 --- a/backend/services/ChunkService/utils/awaitUploadStream.ts +++ b/backend/services/ChunkService/utils/awaitUploadStream.ts @@ -1,31 +1,45 @@ import removeChunks from "../../FileService/utils/removeChunks"; import {Request} from "express" -const awaitUploadStream = (inputSteam: any, outputStream: any, req: Request) => { +const awaitUploadStream = (inputSteam: any, outputStream: any, req: Request, allStreamsToErrorCatch: any[]) => { return new Promise((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", () => { diff --git a/backend/services/ChunkService/utils/awaitUploadStreamFS.ts b/backend/services/ChunkService/utils/awaitUploadStreamFS.ts index 64ec29b..dc8353e 100644 --- a/backend/services/ChunkService/utils/awaitUploadStreamFS.ts +++ b/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 = (inputSteam: any, outputStream: any, req: Request, path: string) => { +const awaitUploadStream = (inputSteam: any, outputStream: any, req: Request, path: string, allStreamsToCatchError: any[]) => { return new Promise((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", () => { diff --git a/backend/utils/streamToBuffer.ts b/backend/utils/streamToBuffer.ts index ec50ee6..edefc02 100644 --- a/backend/utils/streamToBuffer.ts +++ b/backend/utils/streamToBuffer.ts @@ -1,10 +1,27 @@ -const streamToBuffer = (stream: any) => { - const chunks: any[] = [] - return new Promise((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((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; \ No newline at end of file