feat(middleware-compression): compressStream for Node

packages/middleware-compression/src/compressStream.spec.ts

@@ -0,0 +1,46 @@
+import { Readable } from "stream";
+import { createGzip } from "zlib";
+
+import { compressStream } from "./compressStream";
+
+jest.mock("zlib", () => ({
+  createGzip: jest.fn().mockImplementation(() => {
+    const mockStream = new Readable({ read() {} }); // Create a mock readable stream
+    mockStream.pipe = jest.fn().mockReturnValue(mockStream); // Mock the pipe method
+    return mockStream;
+  }),
+}));
+
+describe(compressStream.name, () => {
+  let testDataStream: Readable;
+
+  beforeEach(() => {
+    // Create a readable stream from a string
+    testDataStream = Readable.from("test data");
+  });
+
+  afterEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it("should compress a readble stream using gzip", async () => {
+    const receivedOutput = await compressStream(testDataStream);
+
+    expect(receivedOutput).toBeInstanceOf(Readable);
+    expect(createGzip).toHaveBeenCalledTimes(1);
+  });
+
+  it("should throw an error if compression fails", async () => {
+    const compressionErrorMsg = "compression error message";
+    const compressionError = new Error(compressionErrorMsg);
+    (createGzip as jest.Mock).mockImplementationOnce(() => {
+      throw compressionError;
+    });
+
+    await expect(compressStream(testDataStream)).rejects.toThrow(
+      new Error("Failure during compression: " + compressionErrorMsg)
+    );
+
+    expect(createGzip).toHaveBeenCalledTimes(1);
+  });
+});

packages/middleware-compression/src/compressStream.ts

@@ -0,0 +1,6 @@
+import { Readable } from "stream";
+import { createGzip } from "zlib";
+
+export const compressStream = async (body: Readable): Promise<Readable> => {
+  return body.pipe(createGzip());
+};

packages/middleware-compression/src/compressString.browser.ts

@@ -0,0 +1,29 @@
+import { toUint8Array } from "@smithy/util-utf8";
+
+import { compressStream } from "./compressStream.browser";
+
+export const compressString = async (body: any): Promise<Uint8Array> => {
+  // Only gzip shall be supported initial release.
+
+  const inputUint8Array = toUint8Array(body);
+  const inputStream = new ReadableStream({
+    start(controller) {
+      controller.enqueue(inputUint8Array);
+      controller.close();
+    },
+  });
+
+  const outputStream = await compressStream(inputStream);
+
+  const reader = outputStream.getReader();
+  const chunks = [];
+  while (true) {
+    const { done, value } = await reader.read();
+    if (done) {
+      break;
+    }
+    chunks.push(...value);
+  }
+
+  return new Uint8Array(chunks);
+};