Store decodeStream separately in RestClient class

Signed-off-by: Timothy Johnson <timothy.johnson@broadcom.com>

Author: t1m0thyj

packages/imperative/src/rest/__tests__/client/AbstractRestClient.unit.test.ts

@@ -916,9 +916,10 @@ describe("AbstractRestClient tests", () => {
 
         const fakeResponseStream: any = {
             write: jest.fn(),
-            on: jest.fn(),
-            end: jest.fn((cb: any) => cb()),
-            writableFinished: true
+            on: jest.fn((eventName: string, callback: any) => {
+                if (eventName === "finish") callback();
+            }),
+            end: jest.fn(),
         };
         const fakeRequestStream: any = {
             on: jest.fn((_eventName: string, _callback: any) => {
@@ -1455,7 +1456,7 @@ describe("AbstractRestClient tests", () => {
             }
 
             expect(caughtError instanceof ImperativeError).toBe(true);
-            expect(caughtError.message).toMatchSnapshot();
+            expect(caughtError.message).toContain("Failed to decompress response buffer");
         });
 
         it("should error when decompressing invalid gzip stream", async () => {
@@ -1494,7 +1495,7 @@ describe("AbstractRestClient tests", () => {
             }
 
             expect(caughtError instanceof ImperativeError).toBe(true);
-            expect(caughtError.message).toMatchSnapshot();
+            expect(caughtError.message).toContain("Failed to decompress response stream");
         });
 
         it("should error when decompressing truncated gzip stream with text content", async () => {
@@ -1533,7 +1534,7 @@ describe("AbstractRestClient tests", () => {
             }
 
             expect(caughtError instanceof ImperativeError).toBe(true);
-            expect(caughtError.message).toMatchSnapshot();
+            expect(caughtError.message).toContain("Failed to decompress response stream");
         });
 
         it("should decompress error message for streamed request", async () => {

packages/imperative/src/rest/__tests__/client/__snapshots__/AbstractRestClient.unit.test.ts.snap

@@ -1,11 +1,5 @@
 // Jest Snapshot v1, https://goo.gl/fbAQLP
 
-exports[`AbstractRestClient tests content encoding should error when decompressing invalid gzip buffer 1`] = `"Failed to decompress response buffer with content encoding type gzip"`;
-
-exports[`AbstractRestClient tests content encoding should error when decompressing invalid gzip stream 1`] = `"Failed to decompress response stream with content encoding type gzip"`;
-
-exports[`AbstractRestClient tests content encoding should error when decompressing truncated gzip stream with text content 1`] = `"Failed to decompress response stream with content encoding type gzip"`;
-
 exports[`AbstractRestClient tests should create buildOptions according to input parameter options 1 1`] = `
 Object {
   "headers": Object {

packages/imperative/src/rest/src/client/AbstractRestClient.ts

@@ -183,6 +183,15 @@ export abstract class AbstractRestClient {
      */
     protected mResponseStream: Writable;
 
+    /**
+     * If content encoding is used, this is the decompression stream that we write to.
+     * It pipes through transforms to mResponseStream.
+     * @private
+     * @type {Writable}
+     * @memberof AbstractRestClient
+     */
+    protected mDecodeStream: Writable;
+
     /**
      * stream for outgoing request data to the server
      * @private
@@ -815,7 +824,7 @@ export abstract class AbstractRestClient {
             if (this.mContentEncoding != null) {
                 this.log.debug("Adding decompression transform to response stream");
                 try {
-                    this.mResponseStream = CompressionUtils.decompressStream(this.mResponseStream, this.mContentEncoding,
+                    this.mDecodeStream = CompressionUtils.decompressStream(this.mResponseStream, this.mContentEncoding,
                         this.mNormalizeResponseNewlines);
                 } catch (err) {
                     this.mReject(err);
@@ -878,7 +887,7 @@ export abstract class AbstractRestClient {
                 }
             }
             // write the chunk to the response stream if requested
-            this.mResponseStream.write(respData);
+            (this.mDecodeStream ?? this.mResponseStream).write(respData);
             this.lastByteReceived = respData[respData.byteLength - 1];
         }
     }
@@ -924,7 +933,8 @@ export abstract class AbstractRestClient {
         };
         if (this.mResponseStream != null) {
             this.log.debug("Ending response stream");
-            this.mResponseStream.end(requestEnd);
+            this.mResponseStream.on("finish", requestEnd);
+            (this.mDecodeStream ?? this.mResponseStream).end();
         } else {
             requestEnd();
         }

packages/imperative/src/rest/src/client/CompressionUtils.ts

@@ -9,7 +9,7 @@
 *
 */
 
-import { Duplex, PassThrough, pipeline, Transform, Writable } from "stream";
+import { Transform, Writable } from "stream";
 import * as zlib from "zlib";
 import { ImperativeError } from "../../../error";
 import { IO } from "../../../io";
@@ -51,46 +51,31 @@ export class CompressionUtils {
      * @param normalizeNewLines Specifies if line endings should be converted
      * @throws {ImperativeError}
      */
-    public static decompressStream(responseStream: Writable, encoding: ContentEncoding, normalizeNewLines?: boolean): Duplex {
+    public static decompressStream(responseStream: Writable, encoding: ContentEncoding, normalizeNewLines?: boolean): Writable {
         if (!Headers.CONTENT_ENCODING_TYPES.includes(encoding)) {
             throw new ImperativeError({ msg: `Unsupported content encoding type ${encoding}` });
         }
 
         try {
-            const inputStream = new PassThrough();
+            // First transform handles decompression
+            const transforms = [this.zlibTransform(encoding, !normalizeNewLines)];
 
-            // Build the transform pipeline
-            const transforms: Transform[] = [this.zlibTransform(encoding, !normalizeNewLines)];
+            // Second transform is optional and processes line endings
             if (normalizeNewLines) {
                 transforms.push(this.newLinesTransform());
             }
 
-            // Wrapper stream that waits for pipeline completion
-            let finalCallback: ((error?: Error | null) => void) | undefined;
-            const wrapper = new Duplex({
-                write(chunk, encoding, callback) {
-                    if (!inputStream.write(chunk, encoding)) {
-                        inputStream.once("drain", callback);
-                    } else {
-                        callback();
-                    }
-                },
-                final(callback) {
-                    finalCallback = callback;
-                    inputStream.end();
-                },
-                read() { /* write-only from caller's perspective */ }
-            });
-
-            // Connect pipeline and notify wrapper when complete
-            pipeline([inputStream, ...transforms, responseStream], (err) => {
-                if (err) {
+            // Chain transforms and response stream together
+            for (const [i, stream] of transforms.entries()) {
+                const next = transforms[i + 1] || responseStream;
+                stream.pipe(next);
+                stream.on("error", (err) => {
                     responseStream.emit("error", this.decompressError(err, "stream", encoding));
-                }
-                finalCallback?.(err);
-            });
+                });
+            }
 
-            return wrapper;
+            // Return first stream in chain
+            return transforms[0];
         } catch (err) {
             throw this.decompressError(err, "stream", encoding);
         }