feat(core/protocols): eventstreams for schema serde

Author: kuhe

packages/core/src/submodules/protocols/HttpBindingProtocol.ts

@@ -86,7 +86,7 @@ export abstract class HttpBindingProtocol extends HttpProtocol {
         if (isStreaming) {
           const isEventStream = memberNs.isStructSchema();
           if (isEventStream) {
-            // todo(schema)
+            // todo(schema): serialize event stream payload (REST).
             throw new Error("serialization of event streams is not yet implemented");
           } else {
             // streaming blob body
@@ -223,12 +223,13 @@ export abstract class HttpBindingProtocol extends HttpProtocol {
       }
     }
 
-    const output: Output = {
-      $metadata: this.deserializeMetadata(response),
-      ...dataObject,
-    };
+    const bytes: Uint8Array = await collectBody(response.body, context as SerdeFunctions);
+    if (bytes.byteLength > 0) {
+      Object.assign(dataObject, await deserializer.read(ns, bytes));
+    }
 
-    return output;
+    dataObject.$metadata = this.deserializeMetadata(response);
+    return dataObject;
   }
 
   /**
@@ -277,29 +278,9 @@ export abstract class HttpBindingProtocol extends HttpProtocol {
           const isEventStream = memberSchema.isStructSchema();
           if (isEventStream) {
             // streaming event stream (union)
-            const context = this.serdeContext as unknown as EventStreamSerdeContext;
-            if (!context.eventStreamMarshaller) {
-              throw new Error("@smithy/core - HttpProtocol: eventStreamMarshaller missing in serdeContext.");
-            }
-            const memberSchemas = memberSchema.getMemberSchemas();
-            dataObject[memberName] = context.eventStreamMarshaller.deserialize(response.body, async (event) => {
-              const unionMember =
-                Object.keys(event).find((key) => {
-                  return key !== "__type";
-                }) ?? "";
-              if (unionMember in memberSchemas) {
-                const eventStreamSchema = memberSchemas[unionMember];
-                return {
-                  [unionMember]: await deserializer.read(eventStreamSchema, event[unionMember].body),
-                };
-              } else {
-                // todo(schema): This union convention is ignored by the event stream marshaller.
-                // todo(schema): This should be returned to the user instead.
-                // see "if (deserialized.$unknown) return;" in getUnmarshalledStream.ts
-                return {
-                  $unknown: event,
-                };
-              }
+            dataObject[memberName] = this.deserializeEventStream({
+              response,
+              unionSchema: memberSchema,
             });
           } else {
             // streaming blob body

packages/core/src/submodules/protocols/HttpProtocol.ts

@@ -1,14 +1,19 @@
 import { NormalizedSchema } from "@smithy/core/schema";
+import { SCHEMA } from "@smithy/core/schema";
 import { HttpRequest, HttpResponse } from "@smithy/protocol-http";
-import {
+import type {
   ClientProtocol,
   Codec,
   Endpoint,
   EndpointBearer,
   EndpointV2,
+  EventStreamMarshaller,
+  EventStreamSerdeContext,
   HandlerExecutionContext,
   HttpRequest as IHttpRequest,
   HttpResponse as IHttpResponse,
+  Message as EventStreamMessage,
+  MessageHeaders,
   MetadataBearer,
   OperationSchema,
   ResponseMetadata,
@@ -17,6 +22,7 @@ import {
   ShapeDeserializer,
   ShapeSerializer,
 } from "@smithy/types";
+import { fromUtf8 } from "@smithy/util-utf8";
 
 /**
  * Abstract base for HTTP-based client protocols.
@@ -138,6 +144,78 @@ export abstract class HttpProtocol implements ClientProtocol<IHttpRequest, IHttp
     };
   }
 
+  protected serializeEventStream({ input, unionSchema }: { input: AsyncIterable<any>; unionSchema: NormalizedSchema }) {
+    const marshaller = this.getEventStreamMarshaller();
+    const memberSchemas = unionSchema.getMemberSchemas();
+
+    return marshaller.serialize(input, (event: any): EventStreamMessage => {
+      const unionMember =
+        Object.keys(event).find((key) => {
+          return key !== "__type";
+        }) ?? "";
+      const eventStreamSchema = memberSchemas[unionMember] ?? NormalizedSchema.of(SCHEMA.DOCUMENT);
+
+      this.serializer.write(eventStreamSchema, event);
+      const messageSerialization = this.serializer.flush();
+
+      const body =
+        typeof messageSerialization === "string"
+          ? (this.serdeContext?.utf8Decoder ?? fromUtf8)(messageSerialization)
+          : messageSerialization;
+
+      const headers: MessageHeaders = {
+        ":event-type": { type: "string", value: unionMember },
+        ":message-type": { type: "string", value: "event" },
+        ":content-type": { type: "string", value: this.getDefaultContentType() },
+      };
+
+      return {
+        headers,
+        body,
+      };
+    });
+  }
+
+  protected deserializeEventStream({
+    response,
+    unionSchema,
+  }: {
+    response: IHttpResponse;
+    unionSchema: NormalizedSchema;
+  }): AsyncIterable<{ [key: string]: any; $unknown?: unknown }> {
+    const marshaller = this.getEventStreamMarshaller();
+    const memberSchemas = unionSchema.getMemberSchemas();
+
+    return marshaller.deserialize(response.body, async (event) => {
+      const unionMember =
+        Object.keys(event).find((key) => {
+          return key !== "__type";
+        }) ?? "";
+      if (unionMember in memberSchemas) {
+        const eventStreamSchema = memberSchemas[unionMember];
+        return {
+          [unionMember]: await this.deserializer.read(eventStreamSchema, event[unionMember].body),
+        };
+      } else {
+        // todo(schema): This union convention is ignored by the event stream marshaller.
+        // todo(schema): This should be returned to the user instead.
+        // see "if (deserialized.$unknown) return;" in getUnmarshalledStream.ts
+        return {
+          $unknown: event,
+        };
+      }
+    });
+  }
+
+  /**
+   * @returns content-type default header value for event stream events and other documents.
+   */
+  protected getDefaultContentType(): string {
+    throw new Error(
+      `@smithy/core/protocols - ${this.constructor.name} getDefaultContentType() implementation missing.`
+    );
+  }
+
   /**
    * For HTTP binding protocols, this method is overridden in {@link HttpBindingProtocol}.
    *
@@ -172,4 +250,12 @@ export abstract class HttpProtocol implements ClientProtocol<IHttpRequest, IHttp
     // It should remain unused.
     return [];
   }
+
+  protected getEventStreamMarshaller(): EventStreamMarshaller {
+    const context = this.serdeContext as unknown as EventStreamSerdeContext;
+    if (!context.eventStreamMarshaller) {
+      throw new Error("@smithy/core - HttpProtocol: eventStreamMarshaller missing in serdeContext.");
+    }
+    return context.eventStreamMarshaller;
+  }
 }

packages/core/src/submodules/protocols/RpcProtocol.ts

@@ -56,8 +56,15 @@ export abstract class RpcProtocol extends HttpProtocol {
     };
 
     if (input) {
-      serializer.write(schema, _input);
-      payload = serializer.flush() as Uint8Array;
+      const eventStreamMember = this.getEventStreamMember(ns);
+
+      if (eventStreamMember) {
+        // todo(schema): serialize event stream payload (RPC).
+        throw new Error("serialization of event streams is not yet implemented");
+      } else {
+        serializer.write(schema, _input);
+        payload = serializer.flush() as Uint8Array;
+      }
     }
 
     request.headers = headers;
@@ -93,16 +100,35 @@ export abstract class RpcProtocol extends HttpProtocol {
       response.headers[header.toLowerCase()] = value;
     }
 
-    const bytes: Uint8Array = await collectBody(response.body, context as SerdeFunctions);
-    if (bytes.byteLength > 0) {
-      Object.assign(dataObject, await deserializer.read(ns, bytes));
+    const eventStreamMember = this.getEventStreamMember(ns);
+
+    if (eventStreamMember) {
+      // todo(schema): assign other dataObject fields from initial-response.
+      dataObject[eventStreamMember] = this.deserializeEventStream({
+        response,
+        unionSchema: ns.getMemberSchema(eventStreamMember),
+      });
+    } else {
+      const bytes: Uint8Array = await collectBody(response.body, context as SerdeFunctions);
+      if (bytes.byteLength > 0) {
+        Object.assign(dataObject, await deserializer.read(ns, bytes));
+      }
     }
 
-    const output: Output = {
-      $metadata: this.deserializeMetadata(response),
-      ...dataObject,
-    };
+    dataObject.$metadata = this.deserializeMetadata(response);
+    return dataObject;
+  }
+
+  protected getDefaultContentType(): string {
+    return "application/cbor";
+  }
 
-    return output;
+  protected getEventStreamMember(struct: NormalizedSchema): string {
+    for (const [memberName, memberSchema] of struct.structIterator()) {
+      if (memberSchema.isStreaming() && memberSchema.isStructSchema()) {
+        return memberName;
+      }
+    }
+    return "";
   }
 }