chore: use EventStreamCodec instead of EventStreamMarshaller (#3750)

Author: trivikr

packages/eventstream-handler-node/package.json

@@ -19,7 +19,7 @@
   },
   "license": "Apache-2.0",
   "dependencies": {
-    "@aws-sdk/eventstream-marshaller": "*",
+    "@aws-sdk/eventstream-codec": "*",
     "@aws-sdk/types": "*",
     "tslib": "^2.3.1"
   },

packages/eventstream-handler-node/src/EventSigningStream.spec.ts

@@ -1,16 +1,18 @@
-import { EventStreamMarshaller } from "@aws-sdk/eventstream-marshaller";
+import { EventStreamCodec } from "@aws-sdk/eventstream-codec";
 import { Message, MessageHeaders } from "@aws-sdk/types";
 import { fromUtf8, toUtf8 } from "@aws-sdk/util-utf8-node";
 
 import { EventSigningStream } from "./EventSigningStream";
 
 describe("EventSigningStream", () => {
   const originalDate = Date;
+
   afterEach(() => {
     Date = originalDate;
   });
+
   it("should sign a eventstream payload properly", (done) => {
-    const marshaller = new EventStreamMarshaller(toUtf8, fromUtf8);
+    const eventStreamCodec = new EventStreamCodec(toUtf8, fromUtf8);
     const inputChunks: Array<Uint8Array> = (
       [
         {
@@ -22,7 +24,7 @@ describe("EventSigningStream", () => {
           body: fromUtf8("bar"),
         },
       ] as Array<Message>
-    ).map((event) => marshaller.marshall(event));
+    ).map((event) => eventStreamCodec.encode(event));
     const expected: Array<MessageHeaders> = [
       {
         ":date": { type: "timestamp", value: new Date(1546045446000) },
@@ -57,11 +59,11 @@ describe("EventSigningStream", () => {
     const signingStream = new EventSigningStream({
       priorSignature: "initial",
       eventSigner: { sign: mockEventSigner },
-      eventMarshaller: marshaller,
+      eventStreamCodec,
     });
     const output: Array<MessageHeaders> = [];
     signingStream.on("data", (chunk) => {
-      output.push(marshaller.unmarshall(chunk).headers);
+      output.push(eventStreamCodec.decode(chunk).headers);
     });
     signingStream.on("end", () => {
       expect(output).toEqual(expected);

packages/eventstream-handler-node/src/EventSigningStream.ts

@@ -1,11 +1,11 @@
-import { EventStreamMarshaller as EventMarshaller } from "@aws-sdk/eventstream-marshaller";
+import { EventStreamCodec } from "@aws-sdk/eventstream-codec";
 import { EventSigner, MessageHeaders } from "@aws-sdk/types";
 import { Transform, TransformCallback, TransformOptions } from "stream";
 
 export interface EventSigningStreamOptions extends TransformOptions {
   priorSignature: string;
   eventSigner: EventSigner;
-  eventMarshaller: EventMarshaller;
+  eventStreamCodec: EventStreamCodec;
 }
 
 /**
@@ -14,16 +14,19 @@ export interface EventSigningStreamOptions extends TransformOptions {
 export class EventSigningStream extends Transform {
   private priorSignature: string;
   private eventSigner: EventSigner;
-  private eventMarshaller: EventMarshaller;
+  private eventStreamCodec: EventStreamCodec;
+
   constructor(options: EventSigningStreamOptions) {
     super({
       readableObjectMode: true,
       writableObjectMode: true,
       ...options,
     });
+
     this.priorSignature = options.priorSignature;
     this.eventSigner = options.eventSigner;
-    this.eventMarshaller = options.eventMarshaller;
+    this.eventStreamCodec = options.eventStreamCodec;
+
     //TODO: use 'autoDestroy' when targeting Node 11
     //reference: https://nodejs.org/dist/latest-v13.x/docs/api/stream.html#stream_new_stream_readable_options
     this.on("error", () => {
@@ -43,15 +46,15 @@ export class EventSigningStream extends Transform {
       const signature = await this.eventSigner.sign(
         {
           payload: chunk,
-          headers: this.eventMarshaller.formatHeaders(dateHeader),
+          headers: this.eventStreamCodec.formatHeaders(dateHeader),
         },
         {
           priorSignature: this.priorSignature,
           signingDate: now,
         }
       );
       this.priorSignature = signature;
-      const serializedSigned = this.eventMarshaller.marshall({
+      const serializedSigned = this.eventStreamCodec.encode({
         headers: {
           ...dateHeader,
           ":chunk-signature": {

packages/eventstream-handler-node/src/EventStreamPayloadHandler.spec.ts

@@ -1,13 +1,12 @@
-import { EventStreamMarshaller } from "@aws-sdk/eventstream-marshaller";
+import { EventStreamCodec } from "@aws-sdk/eventstream-codec";
 import { Decoder, Encoder, EventSigner, FinalizeHandler, FinalizeHandlerArguments, HttpRequest } from "@aws-sdk/types";
-import { once } from "events";
 import { PassThrough, Readable } from "stream";
 
 import { EventSigningStream } from "./EventSigningStream";
 import { EventStreamPayloadHandler } from "./EventStreamPayloadHandler";
 
 jest.mock("./EventSigningStream");
-jest.mock("@aws-sdk/eventstream-marshaller");
+jest.mock("@aws-sdk/eventstream-codec");
 
 describe(EventStreamPayloadHandler.name, () => {
   const mockSigner: EventSigner = {
@@ -19,7 +18,7 @@ describe(EventStreamPayloadHandler.name, () => {
 
   beforeEach(() => {
     (EventSigningStream as unknown as jest.Mock).mockImplementation(() => new PassThrough());
-    (EventStreamMarshaller as jest.Mock).mockImplementation(() => {});
+    (EventStreamCodec as jest.Mock).mockImplementation(() => {});
   });
 
   afterEach(() => {
@@ -88,7 +87,7 @@ describe(EventStreamPayloadHandler.name, () => {
     expect(EventSigningStream).toHaveBeenCalledTimes(1);
     expect(EventSigningStream).toHaveBeenCalledWith({
       priorSignature,
-      eventMarshaller: expect.anything(),
+      eventStreamCodec: expect.anything(),
       eventSigner: expect.anything(),
     });
   });

packages/eventstream-handler-node/src/EventStreamPayloadHandler.ts

@@ -1,4 +1,4 @@
-import { EventStreamMarshaller as EventMarshaller } from "@aws-sdk/eventstream-marshaller";
+import { EventStreamCodec } from "@aws-sdk/eventstream-codec";
 import {
   Decoder,
   Encoder,
@@ -31,11 +31,11 @@ export interface EventStreamPayloadHandlerOptions {
  */
 export class EventStreamPayloadHandler implements IEventStreamPayloadHandler {
   private readonly eventSigner: Provider<EventSigner>;
-  private readonly eventMarshaller: EventMarshaller;
+  private readonly eventStreamCodec: EventStreamCodec;
 
   constructor(options: EventStreamPayloadHandlerOptions) {
     this.eventSigner = options.eventSigner;
-    this.eventMarshaller = new EventMarshaller(options.utf8Encoder, options.utf8Decoder);
+    this.eventStreamCodec = new EventStreamCodec(options.utf8Encoder, options.utf8Decoder);
   }
 
   async handle<T extends MetadataBearer>(
@@ -72,7 +72,7 @@ export class EventStreamPayloadHandler implements IEventStreamPayloadHandler {
     const priorSignature = (match || [])[1];
     const signingStream = new EventSigningStream({
       priorSignature,
-      eventMarshaller: this.eventMarshaller,
+      eventStreamCodec: this.eventStreamCodec,
       eventSigner: await this.eventSigner(),
     });
 

packages/eventstream-serde-universal/package.json

@@ -19,7 +19,7 @@
   },
   "license": "Apache-2.0",
   "dependencies": {
-    "@aws-sdk/eventstream-marshaller": "*",
+    "@aws-sdk/eventstream-codec": "*",
     "@aws-sdk/types": "*",
     "tslib": "^2.3.1"
   },

packages/eventstream-serde-universal/src/EventStreamMarshaller.ts

@@ -1,4 +1,4 @@
-import { EventStreamMarshaller as EventMarshaller } from "@aws-sdk/eventstream-marshaller";
+import { EventStreamCodec } from "@aws-sdk/eventstream-codec";
 import { Decoder, Encoder, EventStreamMarshaller as IEventStreamMarshaller, Message } from "@aws-sdk/types";
 
 import { getChunkedStream } from "./getChunkedStream";
@@ -12,10 +12,11 @@ export interface EventStreamMarshallerOptions {
 }
 
 export class EventStreamMarshaller {
-  private readonly eventMarshaller: EventMarshaller;
+  private readonly eventStreamCodec: EventStreamCodec;
   private readonly utfEncoder: Encoder;
+
   constructor({ utf8Encoder, utf8Decoder }: EventStreamMarshallerOptions) {
-    this.eventMarshaller = new EventMarshaller(utf8Encoder, utf8Decoder);
+    this.eventStreamCodec = new EventStreamCodec(utf8Encoder, utf8Decoder);
     this.utfEncoder = utf8Encoder;
   }
 
@@ -25,7 +26,7 @@ export class EventStreamMarshaller {
   ): AsyncIterable<T> {
     const chunkedStream = getChunkedStream(body);
     const unmarshalledStream = getUnmarshalledStream(chunkedStream, {
-      eventMarshaller: this.eventMarshaller,
+      eventStreamCodec: this.eventStreamCodec,
       deserializer,
       toUtf8: this.utfEncoder,
     });
@@ -37,7 +38,7 @@ export class EventStreamMarshaller {
     const self = this;
     const serializedIterator = async function* () {
       for await (const chunk of input) {
-        const payloadBuf = self.eventMarshaller.marshall(serializer(chunk));
+        const payloadBuf = self.eventStreamCodec.encode(serializer(chunk));
         yield payloadBuf;
       }
       // Ending frame

packages/eventstream-serde-universal/src/getUnmarshalledStream.spec.ts

@@ -1,4 +1,4 @@
-import { EventStreamMarshaller } from "@aws-sdk/eventstream-marshaller";
+import { EventStreamCodec } from "@aws-sdk/eventstream-codec";
 import { Message } from "@aws-sdk/types";
 import { fromUtf8, toUtf8 } from "@aws-sdk/util-utf8-node";
 
@@ -53,7 +53,7 @@ describe("getUnmarshalledStream", () => {
       yield endEventMessage;
     };
     const unmarshallerStream = getUnmarshalledStream(source(), {
-      eventMarshaller: new EventStreamMarshaller(toUtf8, fromUtf8),
+      eventStreamCodec: new EventStreamCodec(toUtf8, fromUtf8),
       deserializer: (message) => Promise.resolve(message),
       toUtf8,
     });
@@ -73,7 +73,7 @@ describe("getUnmarshalledStream", () => {
       },
     };
     const deserStream = getUnmarshalledStream(source, {
-      eventMarshaller: new EventStreamMarshaller(toUtf8, fromUtf8),
+      eventStreamCodec: new EventStreamCodec(toUtf8, fromUtf8),
       deserializer: () => {
         throw new Error("error event");
       },
@@ -100,7 +100,7 @@ describe("getUnmarshalledStream", () => {
       },
     };
     const deserStream = getUnmarshalledStream(source, {
-      eventMarshaller: new EventStreamMarshaller(toUtf8, fromUtf8),
+      eventStreamCodec: new EventStreamCodec(toUtf8, fromUtf8),
       deserializer: (message) =>
         Promise.resolve({
           $unknown: message,
@@ -126,7 +126,7 @@ describe("getUnmarshalledStream", () => {
       yield recordEventMessage;
     };
     const unmarshallerStream = getUnmarshalledStream(source(), {
-      eventMarshaller: new EventStreamMarshaller(toUtf8, fromUtf8),
+      eventStreamCodec: new EventStreamCodec(toUtf8, fromUtf8),
       deserializer: (message) =>
         Promise.resolve({
           $unknown: message,

packages/eventstream-serde-universal/src/getUnmarshalledStream.ts

@@ -1,8 +1,8 @@
-import { EventStreamMarshaller as EventMarshaller } from "@aws-sdk/eventstream-marshaller";
+import { EventStreamCodec } from "@aws-sdk/eventstream-codec";
 import { Encoder, Message } from "@aws-sdk/types";
 
 export type UnmarshalledStreamOptions<T> = {
-  eventMarshaller: EventMarshaller;
+  eventStreamCodec: EventStreamCodec;
   deserializer: (input: Record<string, Message>) => Promise<T>;
   toUtf8: Encoder;
 };
@@ -14,7 +14,7 @@ export function getUnmarshalledStream<T extends Record<string, any>>(
   return {
     [Symbol.asyncIterator]: async function* () {
       for await (const chunk of source) {
-        const message = options.eventMarshaller.unmarshall(chunk);
+        const message = options.eventStreamCodec.decode(chunk);
         const { value: messageType } = message.headers[":message-type"];
         if (messageType === "error") {
           // Unmodeled exception in event