chore(eventstream-serde-universal): use EventStreamCodec

Author: trivikr

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