feat(server-common): event stream support

.changeset/beige-dancers-fix.md

@@ -0,0 +1,5 @@
+---
+"@smithy/core": patch
+---
+
+fix for RPC protocol event stream initial messages

.changeset/quiet-tables-lay.md

@@ -0,0 +1,5 @@
+---
+"@smithy/server-node": patch
+---
+
+handle streaming bodies in node-http-converters

.changeset/red-cooks-think.md

@@ -0,0 +1,5 @@
+---
+"@smithy/server-common": minor
+---
+
+event stream support for schema-based server SDK

packages/core/src/submodules/cbor/SmithyRpcV2CborProtocol.ts

@@ -67,9 +67,9 @@ export class SmithyRpcV2CborProtocol extends RpcProtocol {
         this.serializer.write(15, {});
         request.body = this.serializer.flush();
       }
-      try {
-        request.headers["content-length"] = String((request.body as Uint8Array).byteLength);
-      } catch (ignored) {}
+      if (request.body instanceof Uint8Array) {
+        request.headers["content-length"] = String(request.body.byteLength);
+      }
     }
     const { service, operation } = getSmithyContext(context) as {
       service: string;

packages/core/src/submodules/event-streams/EventStreamSerde.spec.ts

@@ -16,6 +16,7 @@ import type {
 import { describe, expect, test as it } from "vitest";
 
 import { EventStreamSerde } from "./EventStreamSerde";
+import { EventStreamCodec } from "./eventstream-codec/EventStreamCodec";
 import { EventStreamMarshaller } from "./eventstream-serde/EventStreamMarshaller";
 
 describe(EventStreamSerde.name, () => {
@@ -763,4 +764,66 @@ describe(EventStreamSerde.name, () => {
       });
     });
   });
+
+  describe("initial-request serialization correctness", () => {
+    it("serializes initial-request string members as strings, not pre-encoded bytes", async () => {
+      // This test verifies that when raw values (not pre-serialized bytes) are
+      // passed as initialRequest, the CBOR serializer correctly encodes them as
+      // their proper types. This catches a regression where pre-serialized
+      // Uint8Array values were passed, causing strings to be double-encoded as blobs.
+      const cborCodec = new CborCodec();
+      const marshaller = new EventStreamMarshaller({ utf8Encoder: toUtf8, utf8Decoder: fromUtf8 });
+      const serde = new EventStreamSerde({
+        marshaller,
+        serializer: cborCodec.createSerializer(),
+        deserializer: cborCodec.createDeserializer(),
+        defaultContentType: "application/cbor",
+      });
+
+      // Schema with a string member and a streaming union member.
+      const streamingUnion: StaticStructureSchema = [
+        4,
+        "ns",
+        "Events",
+        { streaming: 1 },
+        ["alpha"],
+        [[3, "ns", "Alpha", 0, ["id"], [0 satisfies StringSchema]] satisfies StaticStructureSchema],
+      ] as any;
+      const containerSchema: StaticStructureSchema = [
+        3,
+        "ns",
+        "Container",
+        0,
+        ["sessionId", "eventStream"],
+        [0 satisfies StringSchema, [() => streamingUnion, 0]],
+      ];
+
+      const requestBody = await serde.serializeEventStream({
+        eventStream: (async function* () {})(),
+        requestSchema: NormalizedSchema.of(containerSchema),
+        initialRequest: {
+          sessionId: "my-session-value",
+        },
+      });
+
+      // Collect and decode the initial-request message.
+      const chunks: Uint8Array[] = [];
+      for await (const chunk of requestBody as AsyncIterable<Uint8Array>) {
+        chunks.push(chunk);
+      }
+      const fullBody = Buffer.concat(chunks);
+
+      // Read the first message (initial-request).
+      const totalLength = fullBody.readUInt32BE(0);
+      const codec = new EventStreamCodec(toUtf8, fromUtf8);
+      const firstMessage = codec.decode(fullBody.subarray(0, totalLength));
+
+      expect(firstMessage.headers[":event-type"]?.value).toBe("initial-request");
+
+      // Decode the CBOR payload and verify sessionId is a string, not bytes.
+      const decoded = cbor.deserialize(firstMessage.body);
+      expect(typeof decoded.sessionId).toBe("string");
+      expect(decoded.sessionId).toBe("my-session-value");
+    });
+  });
 });

packages/core/src/submodules/event-streams/EventStreamSerde.ts

@@ -67,10 +67,17 @@ export class EventStreamSerde {
     eventStream,
     requestSchema,
     initialRequest,
+    initialMessageType,
   }: {
     eventStream: AsyncIterable<any>;
     requestSchema: NormalizedSchema;
     initialRequest?: any;
+    /**
+     * The :event-type header value for the initial message.
+     * Defaults to "initial-request" (client→server).
+     * Server→client should pass "initial-response".
+     */
+    initialMessageType?: string;
   }): Promise<IHttpRequest["body"] | Uint8Array> {
     const marshaller = this.marshaller;
     const eventStreamMember = requestSchema.getEventStreamMember();
@@ -85,7 +92,7 @@ export class EventStreamSerde {
       async *[Symbol.asyncIterator]() {
         if (initialRequest) {
           const headers: MessageHeaders = {
-            ":event-type": { type: "string", value: "initial-request" },
+            ":event-type": { type: "string", value: initialMessageType ?? "initial-request" },
             ":message-type": { type: "string", value: "event" },
             ":content-type": { type: "string", value: defaultContentType },
           };
@@ -150,10 +157,17 @@ export class EventStreamSerde {
     response,
     responseSchema,
     initialResponseContainer,
+    initialMessageType,
   }: {
     response: IHttpResponse;
     responseSchema: NormalizedSchema;
     initialResponseContainer?: any;
+    /**
+     * The :event-type header value to match as the initial message.
+     * Defaults to "initial-response" (server→client).
+     * Server-side deserialization of client requests should pass "initial-request".
+     */
+    initialMessageType?: string;
   }): Promise<AsyncIterable<{ [key: string]: any; $unknown?: unknown }>> {
     const marshaller = this.marshaller;
     const eventStreamMember = responseSchema.getEventStreamMember();
@@ -173,7 +187,7 @@ export class EventStreamSerde {
 
       const body = event[unionMember].body;
 
-      if (unionMember === "initial-response") {
+      if (unionMember === (initialMessageType ?? "initial-response")) {
         const dataObject = await this.deserializer.read(responseSchema, body);
         delete dataObject[eventStreamMember];
         return {

packages/core/src/submodules/event-streams/eventstream-serde-universal/eventstream-cbor.integ.spec.ts

@@ -1,6 +1,7 @@
 // oxlint-disable no-useless-spread
 import { Readable } from "node:stream";
 import { cbor, dateToTag } from "@smithy/core/cbor";
+import { EventStreamCodec } from "../eventstream-codec/EventStreamCodec";
 import { HttpResponse } from "@smithy/core/protocols";
 import { requireRequestsFrom } from "@smithy/util-test/src";
 import { describe, expect, test as it } from "vitest";
@@ -28,6 +29,10 @@ describe("local model integration test for cbor eventstreams", () => {
       return [...uint32];
     }
 
+    const toUtf8 = (input: Uint8Array): string => new TextDecoder().decode(input);
+    const fromUtf8 = (input: string): Uint8Array => new TextEncoder().encode(input);
+    const codec = new EventStreamCodec(toUtf8, fromUtf8);
+
     requireRequestsFrom(client)
       .toMatch({
         hostname: /localhost/,
@@ -37,25 +42,30 @@ describe("local model integration test for cbor eventstreams", () => {
             outgoing.push(chunk);
           }
           expect(outgoing).toEqual([
-            new Uint8Array([
-              0, 0, 0, 101, 0, 0, 0, 75, 213, 254, 191, 76, 11, 58, 101, 118, 101, 110, 116, 45, 116, 121, 112, 101, 7,
-              0, 5, 97, 108, 112, 104, 97, 13, 58, 109, 101, 115, 115, 97, 103, 101, 45, 116, 121, 112, 101, 7, 0, 5,
-              101, 118, 101, 110, 116, 13, 58, 99, 111, 110, 116, 101, 110, 116, 45, 116, 121, 112, 101, 7, 0, 16, 97,
-              112, 112, 108, 105, 99, 97, 116, 105, 111, 110, 47, 99, 98, 111, 114, 161, 98, 105, 100, 101, 97, 108,
-              112, 104, 97, 32, 93, 69, 236,
-            ]),
-            new Uint8Array([
-              0, 0, 0, 91, 0, 0, 0, 74, 188, 232, 137, 61, 11, 58, 101, 118, 101, 110, 116, 45, 116, 121, 112, 101, 7,
-              0, 4, 98, 101, 116, 97, 13, 58, 109, 101, 115, 115, 97, 103, 101, 45, 116, 121, 112, 101, 7, 0, 5, 101,
-              118, 101, 110, 116, 13, 58, 99, 111, 110, 116, 101, 110, 116, 45, 116, 121, 112, 101, 7, 0, 16, 97, 112,
-              112, 108, 105, 99, 97, 116, 105, 111, 110, 47, 99, 98, 111, 114, 160, 195, 209, 62, 47,
-            ]),
-            new Uint8Array([
-              0, 0, 0, 91, 0, 0, 0, 74, 188, 232, 137, 61, 11, 58, 101, 118, 101, 110, 116, 45, 116, 121, 112, 101, 7,
-              0, 4, 98, 101, 116, 97, 13, 58, 109, 101, 115, 115, 97, 103, 101, 45, 116, 121, 112, 101, 7, 0, 5, 101,
-              118, 101, 110, 116, 13, 58, 99, 111, 110, 116, 101, 110, 116, 45, 116, 121, 112, 101, 7, 0, 16, 97, 112,
-              112, 108, 105, 99, 97, 116, 105, 111, 110, 47, 99, 98, 111, 114, 160, 195, 209, 62, 47,
-            ]),
+            codec.encode({
+              headers: {
+                ":event-type": { type: "string", value: "alpha" },
+                ":message-type": { type: "string", value: "event" },
+                ":content-type": { type: "string", value: "application/cbor" },
+              },
+              body: cbor.serialize({ id: "alpha" }),
+            }),
+            codec.encode({
+              headers: {
+                ":event-type": { type: "string", value: "beta" },
+                ":message-type": { type: "string", value: "event" },
+                ":content-type": { type: "string", value: "application/cbor" },
+              },
+              body: cbor.serialize({}),
+            }),
+            codec.encode({
+              headers: {
+                ":event-type": { type: "string", value: "gamma" },
+                ":message-type": { type: "string", value: "event" },
+                ":content-type": { type: "string", value: "application/cbor" },
+              },
+              body: new Uint8Array(),
+            }),
             new Uint8Array(),
           ]);
         },

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

@@ -64,10 +64,9 @@ export abstract class RpcProtocol extends HttpProtocol {
       if (eventStreamMember) {
         if (input[eventStreamMember]) {
           const initialRequest = {} as any;
-          for (const [memberName, memberSchema] of ns.structIterator()) {
-            if (memberName !== eventStreamMember && input[memberName]) {
-              serializer.write(memberSchema, input[memberName]);
-              initialRequest[memberName] = serializer.flush();
+          for (const [memberName] of ns.structIterator()) {
+            if (memberName !== eventStreamMember && input[memberName] != null) {
+              initialRequest[memberName] = input[memberName];
             }
           }
 

packages/core/src/submodules/protocols/middleware-content-length/contentLengthMiddleware.ts

@@ -26,10 +26,12 @@ export function contentLengthMiddleware(bodyLengthChecker: BodyLengthCalculator)
         ) {
           try {
             const length = bodyLengthChecker(body);
-            request.headers = {
-              ...request.headers,
-              [CONTENT_LENGTH_HEADER]: String(length),
-            };
+            if (length != null) {
+              request.headers = {
+                ...request.headers,
+                [CONTENT_LENGTH_HEADER]: String(length),
+              };
+            }
           } catch (ignored) {
             // ToDo: Add 'transfer-encoding' as chunked only for HTTP/1.1 request
             // Refs: https://github.com/aws/aws-sdk-js-v3/pull/3403