fix(hash-stream-node): throw error if non-file readableStream is flowing (#3341)

Author: trivikr

packages/hash-stream-node/src/readableStreamHasher.spec.ts

@@ -53,11 +53,7 @@ describe(readableStreamHasher.name, () => {
   });
 
   it("creates a copy in case of fileStream", () => {
-    (fsCreateReadStream as jest.Mock).mockReturnValue(
-      new Readable({
-        read: (size) => {},
-      })
-    );
+    (fsCreateReadStream as jest.Mock).mockReturnValue(new Readable({ read: (size) => {} }));
     (isFileStream as unknown as jest.Mock).mockReturnValue(true);
 
     const fsReadStream = createReadStream(__filename);
@@ -68,9 +64,7 @@ describe(readableStreamHasher.name, () => {
   });
 
   it("computes hash for a readable stream", async () => {
-    const readableStream = new Readable({
-      read: (size) => {},
-    });
+    const readableStream = new Readable({ read: (size) => {} });
     const hashPromise = readableStreamHasher(mockHashCtor, readableStream);
 
     // @ts-ignore Property '_readableState' does not exist on type 'Readable'.
@@ -92,6 +86,20 @@ describe(readableStreamHasher.name, () => {
     expect(mockHashCalculatorEnd).toHaveBeenCalledTimes(1);
   });
 
+  it("throws if readable stream is not a file stream and has started reading", async () => {
+    const readableStream = new Readable({ read: (size) => {} });
+    // Simulate readableFlowing to true.
+    readableStream.resume();
+
+    const expectedError = new Error("Unable to calculate hash for flowing readable stream");
+    try {
+      readableStreamHasher(mockHashCtor, readableStream);
+      fail(`expected ${expectedError}`);
+    } catch (error) {
+      expect(error).toStrictEqual(expectedError);
+    }
+  });
+
   it("throws error if readable stream throws error", async () => {
     const readableStream = new Readable({
       read: (size) => {},

packages/hash-stream-node/src/readableStreamHasher.ts

@@ -6,7 +6,11 @@ import { HashCalculator } from "./HashCalculator";
 import { isFileStream } from "./isFileStream";
 
 export const readableStreamHasher: StreamHasher<Readable> = (hashCtor: HashConstructor, readableStream: Readable) => {
-  // ToDo: throw if readableStream is already flowing and it's copy can't be created.
+  // Throw if readableStream is already flowing and it's not a file stream.
+  if (!isFileStream(readableStream) && readableStream.readableFlowing !== null) {
+    throw new Error("Unable to calculate hash for flowing readable stream");
+  }
+
   const streamToPipe = isFileStream(readableStream) ? fsCreateReadStream(readableStream) : readableStream;
 
   const hash = new hashCtor();