feat(core/protocols): eventstreams for schema serde

kuhe · kuhe · commit 8aa092101606 · 2025-08-07T13:12:56.000-04:00
diff --git a/.changeset/shy-hornets-care.md b/.changeset/shy-hornets-care.md
@@ -0,0 +1,5 @@
+---
+"@smithy/core": minor
+---
+
+schema serde eventstreams implementation
diff --git a/packages/core/src/submodules/cbor/SmithyRpcV2CborProtocol.ts b/packages/core/src/submodules/cbor/SmithyRpcV2CborProtocol.ts
@@ -43,9 +43,9 @@ export class SmithyRpcV2CborProtocol extends RpcProtocol {
   ): Promise<IHttpRequest> {
     const request = await super.serializeRequest(operationSchema, input, context);
     Object.assign(request.headers, {
-      "content-type": "application/cbor",
+      "content-type": this.getDefaultContentType(),
       "smithy-protocol": "rpc-v2-cbor",
-      accept: "application/cbor",
+      accept: this.getDefaultContentType(),
     });
     if (deref(operationSchema.input) === "unit") {
       delete request.body;
@@ -113,4 +113,8 @@ export class SmithyRpcV2CborProtocol extends RpcProtocol {
 
     throw exception;
   }
+
+  protected getDefaultContentType(): string {
+    return "application/cbor";
+  }
 }
diff --git a/packages/core/src/submodules/protocols/HttpBindingProtocol.ts b/packages/core/src/submodules/protocols/HttpBindingProtocol.ts
@@ -4,7 +4,6 @@ import { HttpRequest } from "@smithy/protocol-http";
 import {
   Endpoint,
   EndpointBearer,
-  EventStreamSerdeContext,
   HandlerExecutionContext,
   HttpRequest as IHttpRequest,
   HttpResponse as IHttpResponse,
@@ -86,8 +85,12 @@ export abstract class HttpBindingProtocol extends HttpProtocol {
         if (isStreaming) {
           const isEventStream = memberNs.isStructSchema();
           if (isEventStream) {
-            // todo(schema)
-            throw new Error("serialization of event streams is not yet implemented");
+            if (input[memberName]) {
+              payload = this.serializeEventStream({
+                eventStream: input[memberName],
+                unionSchema: memberNs,
+              });
+            }
           } else {
             // streaming blob body
             payload = inputMemberValue;
@@ -132,6 +135,9 @@ export abstract class HttpBindingProtocol extends HttpProtocol {
     if (hasNonHttpBindingMember && input) {
       serializer.write(schema, input);
       payload = serializer.flush() as Uint8Array;
+
+      // Due to Smithy validation, we can assume that the members with no HTTP
+      // bindings DO NOT contain an event stream.
     }
 
     request.headers = headers;
@@ -221,14 +227,12 @@ export abstract class HttpBindingProtocol extends HttpProtocol {
           dataObject[member] = dataFromBody[member];
         }
       }
+      // Due to Smithy validation, we can assume that the members with no HTTP
+      // bindings DO NOT contain an event stream.
     }
 
-    const output: Output = {
-      $metadata: this.deserializeMetadata(response),
-      ...dataObject,
-    };
-
-    return output;
+    dataObject.$metadata = this.deserializeMetadata(response);
+    return dataObject;
   }
 
   /**
@@ -276,33 +280,13 @@ export abstract class HttpBindingProtocol extends HttpProtocol {
         if (isStreaming) {
           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,
-                };
-              }
+            // event stream (union)
+            dataObject[memberName] = this.deserializeEventStream({
+              response,
+              unionSchema: memberSchema,
             });
           } else {
-            // streaming blob body
+            // data stream (blob)
             dataObject[memberName] = sdkStreamMixin(response.body);
           }
         } else if (response.body) {
diff --git a/packages/core/src/submodules/protocols/HttpProtocol.ts b/packages/core/src/submodules/protocols/HttpProtocol.ts
@@ -1,14 +1,20 @@
 import { NormalizedSchema } from "@smithy/core/schema";
+import { SCHEMA } from "@smithy/core/schema";
 import { HttpRequest, HttpResponse } from "@smithy/protocol-http";
 import {
   ClientProtocol,
   Codec,
   Endpoint,
   EndpointBearer,
   EndpointV2,
+  EventStreamMarshaller,
+  EventStreamSerdeContext,
   HandlerExecutionContext,
   HttpRequest as IHttpRequest,
   HttpResponse as IHttpResponse,
+  Message as EventStreamMessage,
+  MessageHeaders,
+  MessageHeaderValue,
   MetadataBearer,
   OperationSchema,
   ResponseMetadata,
@@ -17,6 +23,7 @@ import {
   ShapeDeserializer,
   ShapeSerializer,
 } from "@smithy/types";
+import { fromUtf8 } from "@smithy/util-utf8";
 
 /**
  * Abstract base for HTTP-based client protocols.
@@ -138,6 +145,114 @@ export abstract class HttpProtocol implements ClientProtocol<IHttpRequest, IHttp
     };
   }
 
+  protected serializeEventStream({
+    eventStream,
+    unionSchema,
+  }: {
+    eventStream: AsyncIterable<any>;
+    unionSchema: NormalizedSchema;
+  }) {
+    const marshaller = this.getEventStreamMarshaller();
+    const memberSchemas = unionSchema.getMemberSchemas();
+
+    return marshaller.serialize(eventStream, (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() },
+      };
+
+      // additional trait-annotated event headers.
+      for (const [memberName, memberSchema] of eventStreamSchema.structIterator()) {
+        const isHeader = !!memberSchema.getMergedTraits().eventHeader;
+        if (!isHeader) {
+          continue;
+        }
+        const value = event[memberName];
+        let type = "binary" as MessageHeaderValue["type"];
+        if (memberSchema.isNumericSchema()) {
+          if ((-2) ** 31 <= value && value <= 2 ** 31 - 1) {
+            type = "integer";
+          } else {
+            type = "long";
+          }
+        } else if (memberSchema.isTimestampSchema()) {
+          type = "timestamp";
+        } else if (memberSchema.isStringSchema()) {
+          type = "string";
+        } else if (memberSchema.isBooleanSchema()) {
+          type = "boolean";
+        }
+
+        if (isHeader && value != undefined) {
+          headers[memberName] = {
+            type,
+            value,
+          };
+        }
+      }
+
+      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 +287,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;
+  }
 }
diff --git a/packages/core/src/submodules/protocols/RpcProtocol.ts b/packages/core/src/submodules/protocols/RpcProtocol.ts
@@ -1,6 +1,6 @@
 import { NormalizedSchema, SCHEMA } from "@smithy/core/schema";
 import { HttpRequest } from "@smithy/protocol-http";
-import {
+import type {
   Endpoint,
   EndpointBearer,
   HandlerExecutionContext,
@@ -56,8 +56,19 @@ export abstract class RpcProtocol extends HttpProtocol {
     };
 
     if (input) {
-      serializer.write(schema, _input);
-      payload = serializer.flush() as Uint8Array;
+      const eventStreamMember = ns.getEventStreamMember();
+
+      if (eventStreamMember) {
+        if (_input[eventStreamMember]) {
+          payload = this.serializeEventStream({
+            eventStream: _input[eventStreamMember],
+            unionSchema: ns.getMemberSchema(eventStreamMember),
+          });
+        }
+      } else {
+        serializer.write(schema, _input);
+        payload = serializer.flush() as Uint8Array;
+      }
     }
 
     request.headers = headers;
@@ -93,16 +104,21 @@ 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 = ns.getEventStreamMember();
+    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,
-    };
-
-    return output;
+    dataObject.$metadata = this.deserializeMetadata(response);
+    return dataObject;
   }
 }
diff --git a/packages/core/src/submodules/schema/schemas/NormalizedSchema.spec.ts b/packages/core/src/submodules/schema/schemas/NormalizedSchema.spec.ts
@@ -239,4 +239,39 @@ describe(NormalizedSchema.name, () => {
       }
     });
   });
+
+  describe("event stream detection", () => {
+    it("should retrieve the event stream member", () => {
+      const schema = struct(
+        "ns",
+        "StructureWithEventStream",
+        0,
+        ["A", "B", "C", "D", "EventStream"],
+        [0, 0, 0, 0, struct("ns", "Union", { streaming: 1 }, [], [])]
+      );
+      const ns = NormalizedSchema.of(schema);
+
+      expect(ns.getEventStreamMember()).toEqual("EventStream");
+    });
+
+    it("should return empty string if no event stream member is present", () => {
+      const schema = struct(
+        "ns",
+        "StructureWithEventStream",
+        0,
+        ["A", "B", "C", "D", "EventStream"],
+        [0, 0, 0, 0, struct("ns", "Union", 0, [], [])]
+      );
+      const ns = NormalizedSchema.of(schema);
+
+      expect(ns.getEventStreamMember()).toEqual("");
+    });
+
+    it("should not throw an exception if the NormalizedSchema is not a structure", () => {
+      const schema = 0;
+      const ns = NormalizedSchema.of(schema);
+
+      expect(ns.getEventStreamMember()).toEqual("");
+    });
+  });
 });
diff --git a/packages/core/src/submodules/schema/schemas/NormalizedSchema.ts b/packages/core/src/submodules/schema/schemas/NormalizedSchema.ts
diff --git a/packages/types/src/schema/schema.ts b/packages/types/src/schema/schema.ts
diff --git a/smithy-typescript-codegen/src/main/java/software/amazon/smithy/typescript/codegen/schema/SchemaTraitFilterIndex.java b/smithy-typescript-codegen/src/main/java/software/amazon/smithy/typescript/codegen/schema/SchemaTraitFilterIndex.java

-Original file line number
+Diff line change
@@ @@ -0,0 +1,5 @@ @@
 +---
 +"@smithy/core": minor
 +---
++
 +schema serde eventstreams implementation