fix: return readable stream with async iterators properly (#956)

aws · Mar 2, 2020 · 76553cf · 76553cf
1 parent dde5a6f
commit 76553cf
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 7 deletions.
diff --git a/packages/eventstream-serde-node/src/EventStreamMarshaller.ts b/packages/eventstream-serde-node/src/EventStreamMarshaller.ts
@@ -6,7 +6,7 @@ import {
   EventStreamMarshaller as IEventStreamMarshaller
 } from "@aws-sdk/types";
 import { Readable, pipeline } from "stream";
-import { ReadabletoIterable } from "./utils";
+import { readabletoIterable } from "./utils";
 import { EventMessageChunkerStream } from "./EventMessageChunkerStream";
 import { MessageUnmarshallerStream } from "./MessageUnmarshallerStream";
 import { EventDeserializerStream } from "./EventDeserializerStream";
@@ -44,7 +44,11 @@ export class EventStreamMarshaller {
     );
     //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
-    return ReadabletoIterable(eventDeserializerStream);
+    if (typeof eventDeserializerStream[Symbol.asyncIterator] === "function") {
+      // use the experimental feature if available.
+      return eventDeserializerStream;
+    }
+    return readabletoIterable(eventDeserializerStream);
   }
 
   serialize<T>(

diff --git a/packages/eventstream-serde-node/src/utils.ts b/packages/eventstream-serde-node/src/utils.ts
@@ -18,13 +18,9 @@ export function getSignatureBinary(signature: string): Uint8Array {
  * Reference: https://nodejs.org/docs/latest-v11.x/api/stream.html#stream_readable_symbol_asynciterator
  */
 
-export async function* ReadabletoIterable<T>(
+export async function* readabletoIterable<T>(
   readStream: Readable
 ): AsyncIterable<T> {
-  if (typeof readStream[Symbol.asyncIterator] === "function") {
-    // use the experimental feature if available.
-    throw readStream;
-  }
   let streamEnded = false;
   let generationEnded = false;
   const records = new Array<T>();