feat: handle unmodeled exceptions in eventstream serde

Author: AllanZhengYP

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

@@ -24,8 +24,8 @@ export class EventStreamMarshaller {
   }
 
   deserialize<T>(
-    body: ReadableStream,
-    deserializer: (input: { [event: string]: Message }) => T
+    body: ReadableStream<Uint8Array>,
+    deserializer: (input: { [event: string]: Message }) => Promise<T>
   ): AsyncIterable<T> {
     const chunkedStream = getChunkedStream(body);
     const messageStream = getEventMessageStream(

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

@@ -29,7 +29,7 @@ export class EventStreamMarshaller {
 
   deserialize<T>(
     body: Readable,
-    deserializer: (input: { [event: string]: Message }) => T
+    deserializer: (input: { [event: string]: Message }) => Promise<T>
   ): AsyncIterable<T> {
     //should use stream[Symbol.asyncIterable] when the api is stable
     //reference: https://nodejs.org/docs/latest-v11.x/api/stream.html#stream_readable_symbol_asynciterator

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

@@ -7,7 +7,6 @@ import {
 } from "@aws-sdk/types";
 import { getChunkedStream } from "./getChunkedStream";
 import { getUnmarshalledStream } from "./getUnmarshalledStream";
-import { getDeserializedStream } from "./getDeserializedStream";
 
 export interface EventStreamMarshaller extends IEventStreamMarshaller {}
 
@@ -18,22 +17,23 @@ export interface EventStreamMarshallerOptions {
 
 export class EventStreamMarshaller {
   private readonly eventMarshaller: EventMarshaller;
+  private readonly utfEncoder: Encoder;
   constructor({ utf8Encoder, utf8Decoder }: EventStreamMarshallerOptions) {
     this.eventMarshaller = new EventMarshaller(utf8Encoder, utf8Decoder);
+    this.utfEncoder = utf8Encoder;
   }
 
   deserialize<T>(
     body: AsyncIterable<Uint8Array>,
-    deserializer: (input: { [event: string]: Message }) => T
+    deserializer: (input: { [event: string]: Message }) => Promise<T>
   ): AsyncIterable<T> {
     const chunkedStream = getChunkedStream(body);
     const unmarshalledStream = getUnmarshalledStream(chunkedStream, {
-      eventMarshaller: this.eventMarshaller
+      eventMarshaller: this.eventMarshaller,
+      deserializer,
+      toUtf8: this.utfEncoder
     });
-    const deserializedStream = getDeserializedStream(unmarshalledStream, {
-      deserializer
-    });
-    return deserializedStream;
+    return unmarshalledStream;
   }
 
   serialize<T>(

packages/eventstream-serde-universal/src/fixtures/event.fixture.ts

@@ -12,3 +12,8 @@ export const endEventMessage = Buffer.from(
   "AAAAOAAAACjBxoTUDTptZXNzYWdlLXR5cGUHAAVldmVudAs6ZXZlbnQtdHlwZQcAA0VuZM+X05I=",
   "base64"
 );
+
+export const exception = Buffer.from(
+  "AAAAtgAAAF8BcW64DTpjb250ZW50LXR5cGUHABhhcHBsaWNhdGlvbi9vY3RldC1zdHJlYW0NOm1lc3NhZ2UtdHlwZQcACWV4Y2VwdGlvbg86ZXhjZXB0aW9uLXR5cGUHAAlFeGNlcHRpb25UaGlzIGlzIGEgbW9kZWxlZCBleGNlcHRpb24gZXZlbnQgdGhhdCB3b3VsZCBiZSB0aHJvd24gaW4gZGVzZXJpYWxpemVyLj6Gc60=",
+  "base64"
+);

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

@@ -4,7 +4,8 @@ import { getUnmarshalledStream } from "./getUnmarshalledStream";
 import {
   recordEventMessage,
   statsEventMessage,
-  endEventMessage
+  endEventMessage,
+  exception
 } from "./fixtures/event.fixture";
 import { Message } from "@aws-sdk/types";
 
@@ -56,7 +57,9 @@ describe("getUnmarshalledStream", () => {
       yield endEventMessage;
     };
     const unmarshallerStream = getUnmarshalledStream(source(), {
-      eventMarshaller: new EventStreamMarshaller(toUtf8, fromUtf8)
+      eventMarshaller: new EventStreamMarshaller(toUtf8, fromUtf8),
+      deserializer: message => Promise.resolve(message),
+      toUtf8
     });
     const messages: Array<Message> = [];
     for await (const message of unmarshallerStream) {
@@ -66,4 +69,59 @@ describe("getUnmarshalledStream", () => {
       expect(messages[i]).toEqual(expectedMessages[i]);
     }
   });
+
+  it("throws when deserializer throws an error", async () => {
+    const source = {
+      [Symbol.asyncIterator]: async function* () {
+        yield exception;
+      }
+    };
+    const deserStream = getUnmarshalledStream(source, {
+      eventMarshaller: new EventStreamMarshaller(toUtf8, fromUtf8),
+      deserializer: message => {
+        throw new Error("error event");
+      },
+      toUtf8
+    });
+    let error: Error | undefined = undefined;
+    try {
+      for await (const event of deserStream) {
+        //pass.
+      }
+    } catch (e) {
+      error = e;
+    }
+    expect(error).toBeDefined();
+    expect(error!.message).toEqual("error event");
+  });
+
+  it("throws on exception event if deserializer doesn't throw", async () => {
+    //This could happened if client-side SDK is not up-to-date
+    const source = {
+      [Symbol.asyncIterator]: async function* () {
+        yield exception;
+      }
+    };
+    const deserStream = getUnmarshalledStream(source, {
+      eventMarshaller: new EventStreamMarshaller(toUtf8, fromUtf8),
+      deserializer: message =>
+        Promise.resolve({
+          $unknown: message
+        }),
+      toUtf8
+    });
+    let error: Error | undefined = undefined;
+    try {
+      for await (const event of deserStream) {
+        //pass.
+      }
+    } catch (e) {
+      error = e;
+    }
+    expect(error).toBeDefined();
+    expect(error?.name).toEqual("Exception");
+    expect(error?.message).toEqual(
+      "This is a modeled exception event that would be thrown in deserializer."
+    );
+  });
 });

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

@@ -1,14 +1,16 @@
 import { EventStreamMarshaller as EventMarshaller } from "@aws-sdk/eventstream-marshaller";
-import { Message } from "@aws-sdk/types";
+import { Message, Encoder } from "@aws-sdk/types";
 
-export type UnmarshalledStreamOptions = {
+export type UnmarshalledStreamOptions<T> = {
   eventMarshaller: EventMarshaller;
+  deserializer: (input: { [name: string]: Message }) => Promise<T>;
+  toUtf8: Encoder;
 };
 
-export function getUnmarshalledStream(
+export function getUnmarshalledStream<T extends { [key: string]: any }>(
   source: AsyncIterable<Uint8Array>,
-  options: UnmarshalledStreamOptions
-): AsyncIterable<{ [name: string]: Message }> {
+  options: UnmarshalledStreamOptions<T>
+): AsyncIterable<T> {
   return {
     [Symbol.asyncIterator]: async function* () {
       for await (const chunk of source) {
@@ -24,13 +26,22 @@ export function getUnmarshalledStream(
           throw unmodeledError;
         } else if (messageType === "exception") {
           // For modeled exception, push it to deserializer and throw after deserializing
-          yield {
-            [message.headers[":exception-type"].value as string]: message
-          };
+          const code = message.headers[":exception-type"].value as string;
+          const exception = { [code]: message };
+          // Get parsed exception event in key(error code) value(structured error) pair.
+          const deserializedException = await options.deserializer(exception);
+          if (deserializedException.$unknown) {
+            //this is an unmodeled exception then try parsing it with best effort
+            const error = new Error(options.toUtf8(message.body));
+            error.name = code;
+            throw error;
+          }
+          throw deserializedException[code];
         } else if (messageType === "event") {
-          yield {
+          const event = {
             [message.headers[":event-type"].value as string]: message
           };
+          yield await options.deserializer(event);
         } else {
           throw Error(
             `Unrecognizable event type: ${message.headers[":event-type"].value}`