feat(codegen): experimental SSE (text/event-stream) event stream protocol

.changeset/sse-event-stream-marshaller.md

@@ -0,0 +1,5 @@
+---
+"@smithy/core": minor
+---
+
+add SSE (text/event-stream) event stream marshaller

CONTRIBUTING.md

@@ -31,7 +31,7 @@ under development:
 
 | Experimental Feature | Flag | Description |
 | -------------------- | ---- | ----------- |
-| N/A                  | N/A  | N/A         |
+| SSE protocol         | `experimentalSseProtocol` | Enables the `sseJson` protocol, which frames event streams as Server-Sent Events (`text/event-stream`). |
 
 ## Reporting Bugs/Feature Requests
 

packages/core/src/submodules/event-streams/eventstream-serde-sse/SseEventStreamMarshaller.spec.ts

@@ -0,0 +1,152 @@
+import type { Message } from "@smithy/types";
+import { describe, expect, test as it } from "vitest";
+
+import { SseEventStreamMarshaller } from "./SseEventStreamMarshaller";
+
+const utf8Encoder = (bytes: Uint8Array) => Buffer.from(bytes).toString("utf-8");
+const utf8Decoder = (str: string) => new Uint8Array(Buffer.from(str, "utf-8"));
+
+const marshaller = new SseEventStreamMarshaller({ utf8Encoder, utf8Decoder });
+
+async function* iterate<T>(items: T[]): AsyncIterable<T> {
+  for (const item of items) {
+    yield item;
+  }
+}
+
+async function collect<T>(stream: AsyncIterable<T>): Promise<T[]> {
+  const out: T[] = [];
+  for await (const item of stream) {
+    out.push(item);
+  }
+  return out;
+}
+
+type ChatEvent = { message: { text: string } } | { done: { reason: string } };
+
+const serializer = (event: ChatEvent): Message => {
+  const type = Object.keys(event)[0] as keyof ChatEvent;
+  return {
+    headers: {
+      ":message-type": { type: "string", value: "event" },
+      ":event-type": { type: "string", value: type },
+      ":content-type": { type: "string", value: "application/json" },
+    },
+    body: utf8Decoder(JSON.stringify((event as any)[type])),
+  };
+};
+
+const deserializer = async (input: Record<string, Message>): Promise<any> => {
+  const type = Object.keys(input)[0];
+  if (type === "message" || type === "done") {
+    return { [type]: JSON.parse(utf8Encoder(input[type].body)) };
+  }
+  return { $unknown: [type, input[type]] };
+};
+
+describe("SseEventStreamMarshaller", () => {
+  it("serializes events into text/event-stream frames", async () => {
+    const frames = await collect(
+      marshaller.serialize(iterate<ChatEvent>([{ message: { text: "hi" } }, { done: { reason: "stop" } }]), serializer)
+    );
+    const wire = frames.map(utf8Encoder).join("");
+    expect(wire).toBe(
+      `event: message\ndata: {"text":"hi"}\n\n` + `event: done\ndata: {"reason":"stop"}\n\n`
+    );
+  });
+
+  it("round-trips events through serialize then deserialize", async () => {
+    const events: ChatEvent[] = [{ message: { text: "one" } }, { message: { text: "two" } }, { done: { reason: "eof" } }];
+    const wire = marshaller.serialize(iterate(events), serializer);
+    const back = await collect(marshaller.deserialize(wire, deserializer));
+    expect(back).toEqual(events);
+  });
+
+  it("handles payloads split across chunk boundaries", async () => {
+    const wire = marshaller.serialize(iterate<ChatEvent>([{ message: { text: "chunked" } }]), serializer);
+    const whole = utf8Encoder((await collect(wire))[0]);
+    async function* byChar(): AsyncIterable<Uint8Array> {
+      for (const ch of whole) {
+        yield utf8Decoder(ch);
+      }
+    }
+    const back = await collect(marshaller.deserialize(byChar(), deserializer));
+    expect(back).toEqual([{ message: { text: "chunked" } }]);
+  });
+
+  it("splits raw-newline payloads across multiple data: lines and rejoins them", async () => {
+    const rawBody = "line1\nline2";
+    const rawSerializer = (): Message => ({
+      headers: {
+        ":message-type": { type: "string", value: "event" },
+        ":event-type": { type: "string", value: "raw" },
+        ":content-type": { type: "string", value: "text/plain" },
+      },
+      body: utf8Decoder(rawBody),
+    });
+    const rawDeserializer = async (input: Record<string, Message>): Promise<any> => {
+      const type = Object.keys(input)[0];
+      return type === "raw" ? { raw: utf8Encoder(input[type].body) } : { $unknown: [type, input[type]] };
+    };
+    const frame = utf8Encoder((await collect(marshaller.serialize(iterate([{} as any]), rawSerializer)))[0]);
+    expect(frame.match(/data: /g)?.length).toBe(2);
+    const back = await collect(
+      marshaller.deserialize(marshaller.serialize(iterate([{} as any]), rawSerializer), rawDeserializer)
+    );
+    expect(back).toEqual([{ raw: rawBody }]);
+  });
+
+  it("round-trips CR and CRLF line terminators as LF", async () => {
+    const rawSerializer = (input: any): Message => ({
+      headers: {
+        ":message-type": { type: "string", value: "event" },
+        ":event-type": { type: "string", value: "raw" },
+        ":content-type": { type: "string", value: "text/plain" },
+      },
+      body: utf8Decoder(input.raw),
+    });
+    const rawDeserializer = async (input: Record<string, Message>): Promise<any> => {
+      const type = Object.keys(input)[0];
+      return type === "raw" ? { raw: utf8Encoder(input[type].body) } : { $unknown: [type, input[type]] };
+    };
+    // SSE data: fields cannot preserve which terminator was used; all rejoin as LF.
+    for (const [body, expected] of [
+      ["a\rb", "a\nb"],
+      ["a\r\nb", "a\nb"],
+      ["a\nb", "a\nb"],
+    ]) {
+      const back = await collect(
+        marshaller.deserialize(marshaller.serialize(iterate([{ raw: body }]), rawSerializer), rawDeserializer)
+      );
+      expect(back).toEqual([{ raw: expected }]);
+    }
+  });
+
+  it("maps modeled exceptions to exception: events and throws on deserialize", async () => {
+    const errSerializer = (): Message => ({
+      headers: {
+        ":message-type": { type: "string", value: "exception" },
+        ":exception-type": { type: "string", value: "ThrottlingError" },
+        ":content-type": { type: "string", value: "application/json" },
+      },
+      body: utf8Decoder(JSON.stringify({ message: "slow down" })),
+    });
+    const errDeserializer = async (input: Record<string, Message>): Promise<any> => {
+      const type = Object.keys(input)[0];
+      if (type === "ThrottlingError") {
+        const parsed = JSON.parse(utf8Encoder(input[type].body));
+        const e = new Error(parsed.message);
+        e.name = "ThrottlingError";
+        return { ThrottlingError: e };
+      }
+      return { $unknown: [type, input[type]] };
+    };
+    const wire = marshaller.serialize(iterate([{} as any]), errSerializer);
+    const frame = utf8Encoder((await collect(wire))[0]);
+    expect(frame.startsWith("event: exception:ThrottlingError\n")).toBe(true);
+
+    await expect(
+      collect(marshaller.deserialize(marshaller.serialize(iterate([{} as any]), errSerializer), errDeserializer))
+    ).rejects.toThrow("slow down");
+  });
+});

packages/core/src/submodules/event-streams/eventstream-serde-sse/SseEventStreamMarshaller.ts

@@ -0,0 +1,179 @@
+import type {
+  Decoder,
+  Encoder,
+  EventStreamMarshaller as IEventStreamMarshaller,
+  EventStreamSerdeProvider,
+  Message,
+} from "@smithy/types";
+
+/**
+ * Options for {@link SseEventStreamMarshaller}. Mirrors the options of the
+ * binary event stream marshaller so the two are interchangeable at the
+ * codegen serde-context seam.
+ *
+ * @internal
+ */
+export interface SseEventStreamMarshallerOptions {
+  utf8Encoder: Encoder;
+  utf8Decoder: Decoder;
+}
+
+const EXCEPTION_PREFIX = "exception:";
+
+/**
+ * An {@link IEventStreamMarshaller} that frames Smithy event stream messages
+ * as Server-Sent Events (`text/event-stream`) instead of the binary
+ * `application/vnd.amazon.eventstream` encoding.
+ *
+ * Wire mapping:
+ * - `:event-type` header -> SSE `event:` field
+ * - message body         -> SSE `data:` field(s)
+ * - modeled exceptions   -> SSE `event: exception:<code>`
+ *
+ * The serializer/deserializer callbacks use the same contract as the binary
+ * marshaller: a `Record<string, Message>` keyed by event type on the way in,
+ * a `Message` with `:event-type` / `:message-type` headers on the way out.
+ *
+ * @internal
+ */
+export class SseEventStreamMarshaller implements IEventStreamMarshaller<AsyncIterable<Uint8Array>> {
+  private readonly utf8Encoder: Encoder;
+  private readonly utf8Decoder: Decoder;
+
+  constructor({ utf8Encoder, utf8Decoder }: SseEventStreamMarshallerOptions) {
+    this.utf8Encoder = utf8Encoder;
+    this.utf8Decoder = utf8Decoder;
+  }
+
+  public serialize<T>(input: AsyncIterable<T>, serializer: (event: T) => Message): AsyncIterable<Uint8Array> {
+    const { utf8Decoder, utf8Encoder } = this;
+    return {
+      [Symbol.asyncIterator]: async function* () {
+        for await (const event of input) {
+          const message = serializer(event);
+          const messageType = String(message.headers[":message-type"]?.value ?? "event");
+          let eventName: string;
+          if (messageType === "exception") {
+            eventName = EXCEPTION_PREFIX + String(message.headers[":exception-type"]?.value ?? "UnknownError");
+          } else {
+            eventName = String(message.headers[":event-type"]?.value ?? "message");
+          }
+          const body = message.body.length ? utf8Encoder(message.body) : "";
+          // Per the SSE spec, payloads containing line terminators (CRLF, CR, or LF)
+          // are sent as repeated data: fields and rejoined with LF when parsed.
+          const dataLines = body
+            .split(/\r\n|\r|\n/)
+            .map((line) => `data: ${line}`)
+            .join("\n");
+          yield utf8Decoder(`event: ${eventName}\n${dataLines}\n\n`);
+        }
+      },
+    };
+  }
+
+  public deserialize<T>(
+    body: AsyncIterable<Uint8Array>,
+    deserializer: (input: Record<string, Message>) => Promise<T>
+  ): AsyncIterable<T> {
+    const { utf8Encoder, utf8Decoder } = this;
+    return {
+      [Symbol.asyncIterator]: async function* () {
+        for await (const sse of parseSseStream(body, utf8Encoder)) {
+          const isException = sse.event.startsWith(EXCEPTION_PREFIX);
+          const type = isException ? sse.event.slice(EXCEPTION_PREFIX.length) : sse.event;
+          const message: Message = {
+            headers: {
+              ":message-type": { type: "string", value: isException ? "exception" : "event" },
+              [isException ? ":exception-type" : ":event-type"]: { type: "string", value: type },
+              ":content-type": { type: "string", value: "application/json" },
+            },
+            body: utf8Decoder(sse.data),
+          };
+          const deserialized: any = await deserializer({ [type]: message });
+          if (isException) {
+            if (deserialized.$unknown) {
+              const error = new Error(sse.data || "UnknownError");
+              error.name = type;
+              throw error;
+            }
+            throw deserialized[type];
+          }
+          if (deserialized.$unknown) {
+            continue;
+          }
+          yield deserialized as T;
+        }
+      },
+    };
+  }
+}
+
+/**
+ * @internal
+ */
+export const sseEventStreamSerdeProvider: EventStreamSerdeProvider = (options: SseEventStreamMarshallerOptions) =>
+  new SseEventStreamMarshaller(options);
+
+interface SseEvent {
+  event: string;
+  data: string;
+}
+
+/**
+ * Incrementally parses a byte stream into SSE events per the WHATWG spec's
+ * field rules: events are delimited by blank lines, `data:` fields accumulate
+ * joined by newlines, comment lines (`:`) and unknown fields are ignored.
+ */
+async function* parseSseStream(source: AsyncIterable<Uint8Array>, toUtf8: Encoder): AsyncIterable<SseEvent> {
+  let buffer = "";
+  for await (const chunk of source) {
+    buffer += toUtf8(chunk);
+    let boundary: number;
+    while ((boundary = findEventBoundary(buffer)) !== -1) {
+      const rawEvent = buffer.slice(0, boundary);
+      buffer = buffer.slice(boundary).replace(/^(\r\n|\n|\r){2}/, "");
+      const parsed = parseSseEvent(rawEvent);
+      if (parsed !== undefined) {
+        yield parsed;
+      }
+    }
+  }
+  const trailing = parseSseEvent(buffer);
+  if (trailing !== undefined) {
+    yield trailing;
+  }
+}
+
+function findEventBoundary(buffer: string): number {
+  const match = buffer.match(/(\r\n\r\n|\n\n|\r\r)/);
+  return match?.index ?? -1;
+}
+
+function parseSseEvent(raw: string): SseEvent | undefined {
+  let event = "message";
+  const data: string[] = [];
+  let sawField = false;
+  for (const line of raw.split(/\r\n|\n|\r/)) {
+    if (line === "" || line.startsWith(":")) {
+      continue;
+    }
+    const colon = line.indexOf(":");
+    const field = colon === -1 ? line : line.slice(0, colon);
+    let value = colon === -1 ? "" : line.slice(colon + 1);
+    if (value.startsWith(" ")) {
+      value = value.slice(1);
+    }
+    if (field === "event") {
+      event = value;
+      sawField = true;
+    } else if (field === "data") {
+      data.push(value);
+      sawField = true;
+    }
+    // id: and retry: fields are valid SSE but carry no Smithy meaning; ignored.
+  }
+  if (!sawField) {
+    return undefined;
+  }
+  return { event, data: data.join("\n") };
+}

packages/core/src/submodules/event-streams/index.browser.ts

@@ -30,6 +30,10 @@ export { EventStreamMarshaller, eventStreamSerdeProvider } from "./eventstream-s
 export type { EventStreamMarshallerOptions } from "./eventstream-serde/EventStreamMarshaller.browser";
 export { readableStreamToIterable, iterableToReadableStream } from "./eventstream-serde/utils";
 
+// SSE (text/event-stream) event stream serde
+export { SseEventStreamMarshaller, sseEventStreamSerdeProvider } from "./eventstream-serde-sse/SseEventStreamMarshaller";
+export type { SseEventStreamMarshallerOptions } from "./eventstream-serde-sse/SseEventStreamMarshaller";
+
 // @smithy/eventstream-serde-universal
 export {
   EventStreamMarshaller as UniversalEventStreamMarshaller,

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

@@ -47,5 +47,9 @@ export type {
   EventStreamSerdeResolvedConfig,
 } from "./eventstream-serde-config-resolver/EventStreamSerdeConfig";
 
+// SSE (text/event-stream) event stream serde
+export { SseEventStreamMarshaller, sseEventStreamSerdeProvider } from "./eventstream-serde-sse/SseEventStreamMarshaller";
+export type { SseEventStreamMarshallerOptions } from "./eventstream-serde-sse/SseEventStreamMarshaller";
+
 // EventStreamSerde
 export { EventStreamSerde } from "./EventStreamSerde";