From db312ca82f0e11832b0d9cb0e30ec6f10a30ab51 Mon Sep 17 00:00:00 2001 From: Josh Elkins Date: Wed, 29 Jul 2026 19:57:46 -0500 Subject: [PATCH 1/2] feat: Schema-based event stream serialization --- .../Serializers/EventContentSerializer.swift | 163 +++++++++++ .../Serializers/EventHeaderSerializer.swift | 132 +++++++++ .../Serializers/EventPayloadSerializer.swift | 151 ++++++++++ .../Serializers/EventStreamSerializer.swift | 167 +++++++++++ .../Serializers/EventUnionSerializer.swift | 235 +++++++++++++++ .../Serializers/UnboundMemberSerializer.swift | 126 ++++++++ test-sdks/Package.swift | 4 + .../EventStreamSerializerTests.swift | 277 ++++++++++++++++++ test-sdks/build.gradle.kts | 1 + test-sdks/model/eventstream.smithy | 126 ++++++++ 10 files changed, 1382 insertions(+) create mode 100644 Sources/SmithyEventStreams/Serializers/EventContentSerializer.swift create mode 100644 Sources/SmithyEventStreams/Serializers/EventHeaderSerializer.swift create mode 100644 Sources/SmithyEventStreams/Serializers/EventPayloadSerializer.swift create mode 100644 Sources/SmithyEventStreams/Serializers/EventStreamSerializer.swift create mode 100644 Sources/SmithyEventStreams/Serializers/EventUnionSerializer.swift create mode 100644 Sources/SmithyEventStreams/Serializers/UnboundMemberSerializer.swift create mode 100644 test-sdks/Tests/SmithyEventStreamsTests/EventStreamSerializerTests.swift create mode 100644 test-sdks/model/eventstream.smithy diff --git a/Sources/SmithyEventStreams/Serializers/EventContentSerializer.swift b/Sources/SmithyEventStreams/Serializers/EventContentSerializer.swift new file mode 100644 index 000000000..3f0e3c9ee --- /dev/null +++ b/Sources/SmithyEventStreams/Serializers/EventContentSerializer.swift @@ -0,0 +1,163 @@ +// +// Copyright Amazon.com Inc. or its affiliates. +// All Rights Reserved. +// +// SPDX-License-Identifier: Apache-2.0 +// + +import struct Foundation.Data +import struct Foundation.Date +@_spi(SchemaBasedSerde) +import class Smithy.EventHeaderTrait +@_spi(SchemaBasedSerde) +import class Smithy.EventPayloadTrait +@_spi(SchemaBasedSerde) +import class Smithy.Schema +import protocol Smithy.SmithyDocument +import struct SmithyEventStreamsAPI.Header +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.Codec +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.SerializableStruct +import struct SmithySerialization.SerializerError +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.ShapeSerializer +@_spi(SchemaBasedSerde) +import typealias SmithySerialization.WriteValueConsumer + +/// Serializes the associated value (an event) of a case of a streaming union to a message. +/// +/// The event's members are bound to the message as follows: +/// - Members marked with the `@eventHeader` trait are written to the message headers. +/// - A member marked with the `@eventPayload` trait, if present, is written to the message payload. +/// - If there is no `@eventPayload` member, the members that are not bound to a header are written +/// to the message payload as a structure, using the codec for the protocol in use. +/// +/// See https://smithy.io/2.0/spec/streaming.html#event-message-serialization +final class EventContentSerializer: ShapeSerializer { + let codec: any Codec + let contentType: String + + /// The headers & payload serialized from the event, available after serialization completes. + private(set) var headers: [Header] = [] + private(set) var payload = Data() + + init(codec: any Codec, contentType: String) { + self.codec = codec + self.contentType = contentType + } + + func writeStruct(_ schema: Schema, _ value: S) throws { + + // The member schema for the union case targets the event structure; its members + // carry the event bindings. + let eventSchema = schema.target ?? schema + + // Write the members that are bound to event headers. + // The header serializer ignores any member that isn't marked with @eventHeader. + let headerSerializer = EventHeaderSerializer() + try value.serializeMembers(eventSchema, headerSerializer) + headers = headerSerializer.headers + + if eventSchema.members.contains(where: { $0.hasTrait(EventPayloadTrait.self) }) { + // Write the single member that is bound to the event payload. + // The payload serializer ignores all other members & provides the content type + // that matches the payload member's type. + let payloadSerializer = EventPayloadSerializer(codec: codec, contentType: contentType) + try value.serializeMembers(eventSchema, payloadSerializer) + payload = payloadSerializer.payload + headers.append(contentsOf: payloadSerializer.headers) + } else if eventSchema.members.contains(where: { !$0.hasTrait(EventHeaderTrait.self) }) { + // There is no @eventPayload member, so write the members that are not bound to a + // header as a structure, using a serializer for the protocol in use. + let payloadSerializer = try codec.makeSerializer() + let unboundValue = UnboundMembers(value: value) + try payloadSerializer.writeStruct(eventSchema, unboundValue) + payload = try payloadSerializer.data + headers.append(Header(name: ":content-type", value: .string(contentType))) + } + } + + func writeList(_ schema: Schema, _ value: [E], _ consumer: WriteValueConsumer) throws { + throw notImplemented + } + + func writeMap(_ schema: Schema, _ value: [String: V], _ consumer: WriteValueConsumer) throws { + throw notImplemented + } + + func writeBoolean(_ schema: Schema, _ value: Bool) throws { + throw notImplemented + } + + func writeByte(_ schema: Schema, _ value: Int8) throws { + throw notImplemented + } + + func writeShort(_ schema: Schema, _ value: Int16) throws { + throw notImplemented + } + + func writeInteger(_ schema: Schema, _ value: Int32) throws { + throw notImplemented + } + + func writeLong(_ schema: Schema, _ value: Int64) throws { + throw notImplemented + } + + func writeFloat(_ schema: Schema, _ value: Float) throws { + throw notImplemented + } + + func writeDouble(_ schema: Schema, _ value: Double) throws { + throw notImplemented + } + + func writeBigInteger(_ schema: Schema, _ value: Int64) throws { + throw notImplemented + } + + func writeBigDecimal(_ schema: Schema, _ value: Double) throws { + throw notImplemented + } + + func writeString(_ schema: Schema, _ value: String) throws { + throw notImplemented + } + + func writeBlob(_ schema: Schema, _ value: Data) throws { + throw notImplemented + } + + func writeTimestamp(_ schema: Schema, _ value: Date) throws { + throw notImplemented + } + + func writeDocument(_ schema: Schema, _ value: any SmithyDocument) throws { + throw notImplemented + } + + func writeNull(_ schema: Schema) throws { + throw notImplemented + } + + var data: Data { + get throws { throw notImplemented } + } + + private var notImplemented: SerializerError { .init("Not implemented") } +} + +/// Wraps an event so that the members bound to event headers are left out of the event payload. +private struct UnboundMembers: SerializableStruct { + let value: S + + func serialize(_ serializer: any ShapeSerializer) throws { + throw SerializerError("Not implemented") + } + + func serializeMembers(_ schema: Schema, _ serializer: any ShapeSerializer) throws { + try value.serializeMembers(schema, UnboundMemberSerializer(base: serializer)) + } +} diff --git a/Sources/SmithyEventStreams/Serializers/EventHeaderSerializer.swift b/Sources/SmithyEventStreams/Serializers/EventHeaderSerializer.swift new file mode 100644 index 000000000..1632616ef --- /dev/null +++ b/Sources/SmithyEventStreams/Serializers/EventHeaderSerializer.swift @@ -0,0 +1,132 @@ +// +// Copyright Amazon.com Inc. or its affiliates. +// All Rights Reserved. +// +// SPDX-License-Identifier: Apache-2.0 +// + +import struct Foundation.Data +import struct Foundation.Date +@_spi(SchemaBasedSerde) +import class Smithy.EventHeaderTrait +@_spi(SchemaBasedSerde) +import class Smithy.Schema +import protocol Smithy.SmithyDocument +import struct SmithyEventStreamsAPI.Header +import enum SmithyEventStreamsAPI.HeaderValue +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.SerializableStruct +import struct SmithySerialization.SerializerError +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.ShapeSerializer +@_spi(SchemaBasedSerde) +import typealias SmithySerialization.WriteValueConsumer + +/// A serializer for event stream data that is bound to event headers. +/// +/// This serializer may be passed all of an event's members; members that are not marked with the +/// `@eventHeader` trait are ignored. A member that is marked with `@eventHeader` but has a type +/// that cannot be bound to an event header will throw a "not implemented" error. +/// +/// After serialization, the headers written by this serializer are available in `headers`. +final class EventHeaderSerializer: ShapeSerializer { + + /// The event headers that were written to this serializer, in the order they were written. + private(set) var headers: [Header] = [] + + func writeStruct(_ schema: Schema, _ value: S) throws { + try skipOrThrow(schema) + } + + func writeList(_ schema: Schema, _ value: [E], _ consumer: WriteValueConsumer) throws { + try skipOrThrow(schema) + } + + func writeMap(_ schema: Schema, _ value: [String: V], _ consumer: WriteValueConsumer) throws { + try skipOrThrow(schema) + } + + func writeBoolean(_ schema: Schema, _ value: Bool) throws { + try append(schema, .bool(value)) + } + + func writeByte(_ schema: Schema, _ value: Int8) throws { + try append(schema, .byte(value)) + } + + func writeShort(_ schema: Schema, _ value: Int16) throws { + try append(schema, .int16(value)) + } + + func writeInteger(_ schema: Schema, _ value: Int32) throws { + try append(schema, .int32(value)) + } + + func writeLong(_ schema: Schema, _ value: Int64) throws { + try append(schema, .int64(value)) + } + + func writeFloat(_ schema: Schema, _ value: Float) throws { + try skipOrThrow(schema) + } + + func writeDouble(_ schema: Schema, _ value: Double) throws { + try skipOrThrow(schema) + } + + func writeBigInteger(_ schema: Schema, _ value: Int64) throws { + try skipOrThrow(schema) + } + + func writeBigDecimal(_ schema: Schema, _ value: Double) throws { + try skipOrThrow(schema) + } + + func writeString(_ schema: Schema, _ value: String) throws { + try append(schema, .string(value)) + } + + func writeBlob(_ schema: Schema, _ value: Data) throws { + try append(schema, .byteArray(value)) + } + + func writeTimestamp(_ schema: Schema, _ value: Date) throws { + try append(schema, .timestamp(value)) + } + + func writeDocument(_ schema: Schema, _ value: any SmithyDocument) throws { + try skipOrThrow(schema) + } + + func writeNull(_ schema: Schema) throws { + // A nil header is simply omitted from the message. + } + + var data: Data { + get throws { throw notImplemented } + } + + // MARK: - Private methods + + /// Appends a header for the passed member schema & value, if the member is bound to an event header. + private func append(_ schema: Schema, _ value: HeaderValue) throws { + guard isEventHeader(schema) else { return } + guard let name = schema.memberName else { + throw SerializerError("Event header must be a structure member. Schema: \(schema.id)") + } + headers.append(Header(name: name, value: value)) + } + + /// Ignores members that are not bound to an event header, and throws for those that are, + /// since a member of this type cannot be written to an event header. + private func skipOrThrow(_ schema: Schema) throws { + guard isEventHeader(schema) else { return } + throw SerializerError("Cannot write type \(schema.type) to an event header. Schema: \(schema.id)") + } + + private func isEventHeader(_ schema: Schema) -> Bool { + schema.hasTrait(EventHeaderTrait.self) + } + + private var notImplemented: SerializerError { .init("Not implemented") } +} diff --git a/Sources/SmithyEventStreams/Serializers/EventPayloadSerializer.swift b/Sources/SmithyEventStreams/Serializers/EventPayloadSerializer.swift new file mode 100644 index 000000000..5bb467ad8 --- /dev/null +++ b/Sources/SmithyEventStreams/Serializers/EventPayloadSerializer.swift @@ -0,0 +1,151 @@ +// +// Copyright Amazon.com Inc. or its affiliates. +// All Rights Reserved. +// +// SPDX-License-Identifier: Apache-2.0 +// + +import struct Foundation.Data +import struct Foundation.Date +@_spi(SchemaBasedSerde) +import class Smithy.EventPayloadTrait +@_spi(SchemaBasedSerde) +import class Smithy.Schema +import protocol Smithy.SmithyDocument +import struct SmithyEventStreamsAPI.Header +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.Codec +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.SerializableStruct +import struct SmithySerialization.SerializerError +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.ShapeSerializer +@_spi(SchemaBasedSerde) +import typealias SmithySerialization.WriteValueConsumer + +/// A serializer for the member of an event that is marked with the `@eventPayload` trait. +/// +/// This serializer may be passed all of an event's members; only the member marked with the +/// `@eventPayload` trait is serialized, all others are ignored. +/// +/// The payload's content type is determined by the type of the payload member: +/// a blob is sent as `application/octet-stream`, a string as `text/plain`, and a structure or +/// union is encoded using the codec for the protocol in use. +/// See https://smithy.io/2.0/spec/streaming.html#eventpayload-trait +final class EventPayloadSerializer: ShapeSerializer { + let codec: any Codec + let contentType: String + + /// The serialized event payload. + private(set) var payload = Data() + + /// The `:content-type` header for the payload, if the payload member was written. + private(set) var headers: [Header] = [] + + init(codec: any Codec, contentType: String) { + self.codec = codec + self.contentType = contentType + } + + func writeStruct(_ schema: Schema, _ value: S) throws { + guard isEventPayload(schema) else { return } + + // Encode the payload structure or union using a serializer for the protocol in use. + // The target schema is used so that the payload is written as the root of the + // payload document, without the member name as a key. + let payloadSerializer = try codec.makeSerializer() + try payloadSerializer.writeStruct(schema.target ?? schema, value) + try setPayload(try payloadSerializer.data, contentType: contentType) + } + + func writeBlob(_ schema: Schema, _ value: Data) throws { + guard isEventPayload(schema) else { return } + try setPayload(value, contentType: "application/octet-stream") + } + + func writeString(_ schema: Schema, _ value: String) throws { + guard isEventPayload(schema) else { return } + try setPayload(Data(value.utf8), contentType: "text/plain") + } + + func writeList(_ schema: Schema, _ value: [E], _ consumer: WriteValueConsumer) throws { + try skipOrThrow(schema) + } + + func writeMap(_ schema: Schema, _ value: [String: V], _ consumer: WriteValueConsumer) throws { + try skipOrThrow(schema) + } + + func writeBoolean(_ schema: Schema, _ value: Bool) throws { + try skipOrThrow(schema) + } + + func writeByte(_ schema: Schema, _ value: Int8) throws { + try skipOrThrow(schema) + } + + func writeShort(_ schema: Schema, _ value: Int16) throws { + try skipOrThrow(schema) + } + + func writeInteger(_ schema: Schema, _ value: Int32) throws { + try skipOrThrow(schema) + } + + func writeLong(_ schema: Schema, _ value: Int64) throws { + try skipOrThrow(schema) + } + + func writeFloat(_ schema: Schema, _ value: Float) throws { + try skipOrThrow(schema) + } + + func writeDouble(_ schema: Schema, _ value: Double) throws { + try skipOrThrow(schema) + } + + func writeBigInteger(_ schema: Schema, _ value: Int64) throws { + try skipOrThrow(schema) + } + + func writeBigDecimal(_ schema: Schema, _ value: Double) throws { + try skipOrThrow(schema) + } + + func writeTimestamp(_ schema: Schema, _ value: Date) throws { + try skipOrThrow(schema) + } + + func writeDocument(_ schema: Schema, _ value: any SmithyDocument) throws { + try skipOrThrow(schema) + } + + func writeNull(_ schema: Schema) throws { + // A nil payload is sent as an empty payload. + } + + var data: Data { + get throws { payload } + } + + // MARK: - Private methods + + private func setPayload(_ payload: Data, contentType: String) throws { + self.payload = payload + self.headers = [Header(name: ":content-type", value: .string(contentType))] + } + + /// Ignores members that are not the event payload, and throws for the payload member if its + /// type may not be bound to an event payload. + private func skipOrThrow(_ schema: Schema) throws { + guard isEventPayload(schema) else { return } + throw SerializerError( + "Expected blob, string, structure, or union for @eventPayload member, " + + "got \(schema.type). Schema: \(schema.id)" + ) + } + + private func isEventPayload(_ schema: Schema) -> Bool { + schema.hasTrait(EventPayloadTrait.self) + } +} diff --git a/Sources/SmithyEventStreams/Serializers/EventStreamSerializer.swift b/Sources/SmithyEventStreams/Serializers/EventStreamSerializer.swift new file mode 100644 index 000000000..9c4ae60b6 --- /dev/null +++ b/Sources/SmithyEventStreams/Serializers/EventStreamSerializer.swift @@ -0,0 +1,167 @@ +// +// Copyright Amazon.com Inc. or its affiliates. +// All Rights Reserved. +// +// SPDX-License-Identifier: Apache-2.0 +// + +import struct Foundation.Data +import struct Foundation.Date +import enum Smithy.ByteStream +@_spi(SchemaBasedSerde) +import class Smithy.Schema +import protocol Smithy.SmithyDocument +@_spi(SchemaBasedSerde) +import class Smithy.StreamingTrait +import typealias SmithyEventStreamsAPI.MarshalClosure +import struct SmithyEventStreamsAPI.Message +import protocol SmithyEventStreamsAPI.MessageEncoder +import protocol SmithyEventStreamsAuthAPI.MessageSigner +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.Codec +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.SerializableStruct +import struct SmithySerialization.SerializerError +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.ShapeSerializer +@_spi(SchemaBasedSerde) +import typealias SmithySerialization.WriteValueConsumer + +/// A serializer that may be used to serialize an event stream to a request. +/// +/// This serializer should only be used on an input structure. It will only serialize the +/// event stream on the input, nothing else. If the input structure has no event stream +/// member, it will throw an error. +/// It will throw a "not implemented" error if serialization of any other type is attempted. +@_spi(SchemaBasedSerde) +public final class EventStreamSerializer: ShapeSerializer { + let codec: any Codec + let contentType: String + let messageEncoder: MessageEncoder + let messageSigner: MessageSigner + let initialRequestMessage: Message? + + /// The request body carrying the encoded, signed event stream, available after serialization. + private(set) public var body: ByteStream = .noStream + + public init( + codec: any Codec, + contentType: String, + messageEncoder: MessageEncoder, + messageSigner: MessageSigner, + initialRequestMessage: Message? = nil + ) { + self.codec = codec + self.contentType = contentType + self.messageEncoder = messageEncoder + self.messageSigner = messageSigner + self.initialRequestMessage = initialRequestMessage + } + + public func writeStruct(_ schema: Schema, _ value: S) throws { + + // Locate the event stream member on this structure. + guard schema.members.contains(where: { $0.type == .union && $0.hasTrait(StreamingTrait.self) }) else { + throw SerializerError("Streaming request sent but no event streaming member") + } + + // Serialize the input's members with this same serializer. + // The writeEventStream method immediately below will be called to serialize the event stream; + // all other members are ignored because they are not part of the event stream body. + try value.serializeMembers(schema, self) + } + + public func writeEventStream( + _ schema: Schema, + _ value: AsyncThrowingStream + ) throws { + + // A marshal closure is created that uses the EventUnionSerializer to marshal each event + // on the stream. This lets us use schema-based serialization with the existing + // marshal/unmarshal interface. + let codec = self.codec + let contentType = self.contentType + let marshalClosure: MarshalClosure = { event in + let eventUnionSerializer = EventUnionSerializer(codec: codec, contentType: contentType) + try event.serialize(eventUnionSerializer) + return eventUnionSerializer.message + } + + // Create a message encoder stream & use that as the body of the request. + body = .stream(DefaultMessageEncoderStream( + stream: value, + messageEncoder: messageEncoder, + marshalClosure: marshalClosure, + messageSigner: messageSigner, + initialRequestMessage: initialRequestMessage + )) + } + + public func writeList(_ schema: Schema, _ value: [E], _ consumer: WriteValueConsumer) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeMap(_ schema: Schema, _ value: [String: V], _ consumer: WriteValueConsumer) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeBoolean(_ schema: Schema, _ value: Bool) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeByte(_ schema: Schema, _ value: Int8) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeShort(_ schema: Schema, _ value: Int16) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeInteger(_ schema: Schema, _ value: Int32) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeLong(_ schema: Schema, _ value: Int64) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeFloat(_ schema: Schema, _ value: Float) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeDouble(_ schema: Schema, _ value: Double) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeBigInteger(_ schema: Schema, _ value: Int64) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeBigDecimal(_ schema: Schema, _ value: Double) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeString(_ schema: Schema, _ value: String) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeBlob(_ schema: Schema, _ value: Data) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeTimestamp(_ schema: Schema, _ value: Date) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeDocument(_ schema: Schema, _ value: any SmithyDocument) throws { + // Members other than the event stream are not part of the event stream body. + } + + public func writeNull(_ schema: Schema) throws { + // Members other than the event stream are not part of the event stream body. + } + + public var data: Data { + get throws { throw SerializerError("Not implemented") } + } +} diff --git a/Sources/SmithyEventStreams/Serializers/EventUnionSerializer.swift b/Sources/SmithyEventStreams/Serializers/EventUnionSerializer.swift new file mode 100644 index 000000000..bddc1e653 --- /dev/null +++ b/Sources/SmithyEventStreams/Serializers/EventUnionSerializer.swift @@ -0,0 +1,235 @@ +// +// Copyright Amazon.com Inc. or its affiliates. +// All Rights Reserved. +// +// SPDX-License-Identifier: Apache-2.0 +// + +import struct Foundation.Data +import struct Foundation.Date +import enum Smithy.ClientError +@_spi(SchemaBasedSerde) +import class Smithy.Schema +import protocol Smithy.SmithyDocument +import struct SmithyEventStreamsAPI.Header +import struct SmithyEventStreamsAPI.Message +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.Codec +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.SerializableStruct +import struct SmithySerialization.SerializerError +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.ShapeSerializer +@_spi(SchemaBasedSerde) +import typealias SmithySerialization.WriteValueConsumer + +/// A serializer that is used to serialize an event stream union to an event stream message. +/// +/// Only use this serializer for event stream unions. Serializing any other type will result in a +/// "not implemented" error. +/// +/// The case of the union that is serialized determines the `:event-type` header, and its associated +/// value is serialized to the message's headers & payload. The `sdkUnknown` case cannot be +/// serialized and results in an error. +final class EventUnionSerializer: ShapeSerializer { + let codec: any Codec + let contentType: String + + /// The message that the union was serialized to, available after serialization completes. + private(set) var message = Message() + + init(codec: any Codec, contentType: String) { + self.codec = codec + self.contentType = contentType + } + + func writeStruct(_ schema: Schema, _ value: S) throws { + + // The union serializes exactly one of its members: the case that is set. + // The event serializer below captures that member & its serialized content. + let eventSerializer = EventSerializer(codec: codec, contentType: contentType) + try value.serializeMembers(schema, eventSerializer) + + guard let eventType = eventSerializer.eventType else { + // The only union case with no member schema is sdkUnknown, which cannot be sent. + throw ClientError.unknownError("cannot serialize the unknown event type!") + } + + var headers = [ + Header(name: ":message-type", value: .string("event")), + Header(name: ":event-type", value: .string(eventType)), + ] + headers.append(contentsOf: eventSerializer.headers) + message = Message(headers: headers, payload: eventSerializer.payload) + } + + func writeList(_ schema: Schema, _ value: [E], _ consumer: WriteValueConsumer) throws { + throw notImplemented + } + + func writeMap(_ schema: Schema, _ value: [String: V], _ consumer: WriteValueConsumer) throws { + throw notImplemented + } + + func writeBoolean(_ schema: Schema, _ value: Bool) throws { + throw notImplemented + } + + func writeByte(_ schema: Schema, _ value: Int8) throws { + throw notImplemented + } + + func writeShort(_ schema: Schema, _ value: Int16) throws { + throw notImplemented + } + + func writeInteger(_ schema: Schema, _ value: Int32) throws { + throw notImplemented + } + + func writeLong(_ schema: Schema, _ value: Int64) throws { + throw notImplemented + } + + func writeFloat(_ schema: Schema, _ value: Float) throws { + throw notImplemented + } + + func writeDouble(_ schema: Schema, _ value: Double) throws { + throw notImplemented + } + + func writeBigInteger(_ schema: Schema, _ value: Int64) throws { + throw notImplemented + } + + func writeBigDecimal(_ schema: Schema, _ value: Double) throws { + throw notImplemented + } + + func writeString(_ schema: Schema, _ value: String) throws { + throw notImplemented + } + + func writeBlob(_ schema: Schema, _ value: Data) throws { + throw notImplemented + } + + func writeTimestamp(_ schema: Schema, _ value: Date) throws { + throw notImplemented + } + + func writeDocument(_ schema: Schema, _ value: any SmithyDocument) throws { + throw notImplemented + } + + func writeNull(_ schema: Schema) throws { + throw notImplemented + } + + var data: Data { + get throws { throw notImplemented } + } + + private var notImplemented: SerializerError { .init("Not implemented") } +} + +/// Serializes the event that is the associated value of the streaming union's selected case. +/// +/// A union writes only the member for the case that is set, so the event type is taken from the +/// name of the single member written to this serializer. +private final class EventSerializer: ShapeSerializer { + let codec: any Codec + let contentType: String + + /// The member name of the union case that was serialized, which names the event type. + private(set) var eventType: String? + private(set) var headers: [Header] = [] + private(set) var payload = Data() + + init(codec: any Codec, contentType: String) { + self.codec = codec + self.contentType = contentType + } + + func writeStruct(_ schema: Schema, _ value: S) throws { + eventType = schema.memberName + + // Serialize the event's members to the message headers & payload. + let contentSerializer = EventContentSerializer(codec: codec, contentType: contentType) + try contentSerializer.writeStruct(schema, value) + headers = contentSerializer.headers + payload = contentSerializer.payload + } + + func writeList(_ schema: Schema, _ value: [E], _ consumer: WriteValueConsumer) throws { + throw notImplemented + } + + func writeMap(_ schema: Schema, _ value: [String: V], _ consumer: WriteValueConsumer) throws { + throw notImplemented + } + + func writeBoolean(_ schema: Schema, _ value: Bool) throws { + throw notImplemented + } + + func writeByte(_ schema: Schema, _ value: Int8) throws { + throw notImplemented + } + + func writeShort(_ schema: Schema, _ value: Int16) throws { + throw notImplemented + } + + func writeInteger(_ schema: Schema, _ value: Int32) throws { + throw notImplemented + } + + func writeLong(_ schema: Schema, _ value: Int64) throws { + throw notImplemented + } + + func writeFloat(_ schema: Schema, _ value: Float) throws { + throw notImplemented + } + + func writeDouble(_ schema: Schema, _ value: Double) throws { + throw notImplemented + } + + func writeBigInteger(_ schema: Schema, _ value: Int64) throws { + throw notImplemented + } + + func writeBigDecimal(_ schema: Schema, _ value: Double) throws { + throw notImplemented + } + + func writeString(_ schema: Schema, _ value: String) throws { + // The sdkUnknown case of a union writes a string; it is not a valid event & is + // detected by the absence of an event type in EventUnionSerializer. + } + + func writeBlob(_ schema: Schema, _ value: Data) throws { + throw notImplemented + } + + func writeTimestamp(_ schema: Schema, _ value: Date) throws { + throw notImplemented + } + + func writeDocument(_ schema: Schema, _ value: any SmithyDocument) throws { + throw notImplemented + } + + func writeNull(_ schema: Schema) throws { + throw notImplemented + } + + var data: Data { + get throws { throw notImplemented } + } + + private var notImplemented: SerializerError { .init("Not implemented") } +} diff --git a/Sources/SmithyEventStreams/Serializers/UnboundMemberSerializer.swift b/Sources/SmithyEventStreams/Serializers/UnboundMemberSerializer.swift new file mode 100644 index 000000000..ccfd6fdd8 --- /dev/null +++ b/Sources/SmithyEventStreams/Serializers/UnboundMemberSerializer.swift @@ -0,0 +1,126 @@ +// +// Copyright Amazon.com Inc. or its affiliates. +// All Rights Reserved. +// +// SPDX-License-Identifier: Apache-2.0 +// + +import struct Foundation.Data +import struct Foundation.Date +@_spi(SchemaBasedSerde) +import class Smithy.EventHeaderTrait +@_spi(SchemaBasedSerde) +import class Smithy.Schema +import protocol Smithy.SmithyDocument +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.SerializableStruct +@_spi(SchemaBasedSerde) +import protocol SmithySerialization.ShapeSerializer +@_spi(SchemaBasedSerde) +import typealias SmithySerialization.WriteValueConsumer + +/// Serializes an event's members to the event payload, omitting the members that are bound +/// to event headers. +/// +/// All writes are forwarded to the serializer for the protocol in use, except for those of +/// members marked with the `@eventHeader` trait, which are dropped because they have already +/// been written to the message's headers. +/// +/// This filtering is only needed for the event's own members; nested shapes within the payload +/// cannot have event bindings, so they are written directly by the underlying serializer. +struct UnboundMemberSerializer: ShapeSerializer { + let base: any ShapeSerializer + + func writeStruct(_ schema: Schema, _ value: S) throws { + guard !isEventHeader(schema) else { return } + try base.writeStruct(schema, value) + } + + func writeList(_ schema: Schema, _ value: [E], _ consumer: WriteValueConsumer) throws { + guard !isEventHeader(schema) else { return } + try base.writeList(schema, value, consumer) + } + + func writeMap(_ schema: Schema, _ value: [String: V], _ consumer: WriteValueConsumer) throws { + guard !isEventHeader(schema) else { return } + try base.writeMap(schema, value, consumer) + } + + func writeBoolean(_ schema: Schema, _ value: Bool) throws { + guard !isEventHeader(schema) else { return } + try base.writeBoolean(schema, value) + } + + func writeByte(_ schema: Schema, _ value: Int8) throws { + guard !isEventHeader(schema) else { return } + try base.writeByte(schema, value) + } + + func writeShort(_ schema: Schema, _ value: Int16) throws { + guard !isEventHeader(schema) else { return } + try base.writeShort(schema, value) + } + + func writeInteger(_ schema: Schema, _ value: Int32) throws { + guard !isEventHeader(schema) else { return } + try base.writeInteger(schema, value) + } + + func writeLong(_ schema: Schema, _ value: Int64) throws { + guard !isEventHeader(schema) else { return } + try base.writeLong(schema, value) + } + + func writeFloat(_ schema: Schema, _ value: Float) throws { + guard !isEventHeader(schema) else { return } + try base.writeFloat(schema, value) + } + + func writeDouble(_ schema: Schema, _ value: Double) throws { + guard !isEventHeader(schema) else { return } + try base.writeDouble(schema, value) + } + + func writeBigInteger(_ schema: Schema, _ value: Int64) throws { + guard !isEventHeader(schema) else { return } + try base.writeBigInteger(schema, value) + } + + func writeBigDecimal(_ schema: Schema, _ value: Double) throws { + guard !isEventHeader(schema) else { return } + try base.writeBigDecimal(schema, value) + } + + func writeString(_ schema: Schema, _ value: String) throws { + guard !isEventHeader(schema) else { return } + try base.writeString(schema, value) + } + + func writeBlob(_ schema: Schema, _ value: Data) throws { + guard !isEventHeader(schema) else { return } + try base.writeBlob(schema, value) + } + + func writeTimestamp(_ schema: Schema, _ value: Date) throws { + guard !isEventHeader(schema) else { return } + try base.writeTimestamp(schema, value) + } + + func writeDocument(_ schema: Schema, _ value: any SmithyDocument) throws { + guard !isEventHeader(schema) else { return } + try base.writeDocument(schema, value) + } + + func writeNull(_ schema: Schema) throws { + guard !isEventHeader(schema) else { return } + try base.writeNull(schema) + } + + var data: Data { + get throws { try base.data } + } + + private func isEventHeader(_ schema: Schema) -> Bool { + schema.hasTrait(EventHeaderTrait.self) + } +} diff --git a/test-sdks/Package.swift b/test-sdks/Package.swift index 99599e0ff..433c7b7aa 100644 --- a/test-sdks/Package.swift +++ b/test-sdks/Package.swift @@ -21,6 +21,7 @@ let package = Package( // Generated test SDKs. Models are in build/model. Use them where Smithy generates them. // Run bash script ./scripts/codegen.sh from smithy-swift root to generate or regenerate these files testSDKPackage("AWSJSON"), + testSDKPackage("EventStream"), testSDKPackage("HTTPLabel"), testSDKPackage("HTTPQuery"), testSDKPackage("JSONName"), @@ -149,6 +150,9 @@ let package = Package( name: "SmithyEventStreamsTests", dependencies: [ .product(name: "SmithyEventStreams", package: "smithy-swift"), + .product(name: "SmithyAWSJSON", package: "smithy-swift"), + .product(name: "SmithySerialization", package: "smithy-swift"), + testSDKProduct("EventStream"), ] ), .testTarget( diff --git a/test-sdks/Tests/SmithyEventStreamsTests/EventStreamSerializerTests.swift b/test-sdks/Tests/SmithyEventStreamsTests/EventStreamSerializerTests.swift new file mode 100644 index 000000000..2125dca0e --- /dev/null +++ b/test-sdks/Tests/SmithyEventStreamsTests/EventStreamSerializerTests.swift @@ -0,0 +1,277 @@ +// +// Copyright Amazon.com Inc. or its affiliates. +// All Rights Reserved. +// +// SPDX-License-Identifier: Apache-2.0 +// + +import XCTest +import struct Foundation.Data +import struct Foundation.Date +@_spi(SchemaBasedSerde) +@_spi(SmithyEventStreams) +import SmithyEventStreams +import enum SmithyEventStreamsAPI.HeaderValue +import struct SmithyEventStreamsAPI.Message +import protocol SmithyEventStreamsAuthAPI.MessageSigner +@_spi(SchemaBasedSerde) +import SmithyAWSJSON +@_spi(SchemaBasedSerde) +import SmithySerialization +@_spi(SchemaBasedSerde) +import EventStreamTestSDK + +/// Tests that the event stream serializers marshal an event stream union to the message +/// format described at https://smithy.io/2.0/spec/streaming.html#event-message-serialization +final class EventStreamSerializerTests: XCTestCase { + typealias TestStream = EventStreamClientTypes.TestEventStream + + // MARK: - @eventPayload bound members + + func test_marshal_writesABlobPayloadAsOctetStream() async throws { + let data = Data("abcdefg".utf8) + let event = TestStream.messagewithblob(.init(data: data)) + + let message = try await marshal(event) + + XCTAssertEqual(headers(of: message), [ + ":message-type": .string("event"), + ":event-type": .string("MessageWithBlob"), + ":content-type": .string("application/octet-stream"), + ]) + XCTAssertEqual(message.payload, data) + } + + func test_marshal_writesAStringPayloadAsTextPlain() async throws { + let event = TestStream.messagewithstring(.init(data: "abcdefg")) + + let message = try await marshal(event) + + XCTAssertEqual(headers(of: message), [ + ":message-type": .string("event"), + ":event-type": .string("MessageWithString"), + ":content-type": .string("text/plain"), + ]) + XCTAssertEqual(message.payload, Data("abcdefg".utf8)) + } + + func test_marshal_writesAStructPayloadUsingTheCodec() async throws { + let event = TestStream.messagewithstruct(.init( + someStruct: .init(someInt: 5, someString: "abc") + )) + + let message = try await marshal(event) + + XCTAssertEqual(headers(of: message), [ + ":message-type": .string("event"), + ":event-type": .string("MessageWithStruct"), + ":content-type": .string("application/json"), + ]) + XCTAssertEqual(try json(message), #"{"someInt":5,"someString":"abc"}"#) + } + + func test_marshal_writesAUnionPayloadUsingTheCodec() async throws { + let event = TestStream.messagewithunion(.init(someUnion: .foo("abc"))) + + let message = try await marshal(event) + + XCTAssertEqual(headers(of: message), [ + ":message-type": .string("event"), + ":event-type": .string("MessageWithUnion"), + ":content-type": .string("application/json"), + ]) + XCTAssertEqual(try json(message), #"{"foo":"abc"}"#) + } + + // MARK: - @eventHeader bound members + + func test_marshal_writesEveryHeaderType() async throws { + let date = Date(timeIntervalSince1970: 1_733_000_000) + let event = TestStream.messagewithheaders(.init( + blob: Data("xyz".utf8), + boolean: true, + byte: 7, + int: 100_000, + long: 5_000_000_000, + short: 300, + string: "abc", + timestamp: date + )) + + let message = try await marshal(event) + + XCTAssertEqual(headers(of: message), [ + ":message-type": .string("event"), + ":event-type": .string("MessageWithHeaders"), + "blob": .byteArray(Data("xyz".utf8)), + "boolean": .bool(true), + "byte": .byte(7), + "int": .int32(100_000), + "long": .int64(5_000_000_000), + "short": .int16(300), + "string": .string("abc"), + "timestamp": .timestamp(date), + ]) + + // An event with only header-bound members has no payload, and so no content type. + XCTAssertEqual(message.payload, Data()) + } + + func test_marshal_omitsNilHeaders() async throws { + let event = TestStream.messagewithheaders(.init(string: "abc")) + + let message = try await marshal(event) + + XCTAssertEqual(headers(of: message), [ + ":message-type": .string("event"), + ":event-type": .string("MessageWithHeaders"), + "string": .string("abc"), + ]) + } + + func test_marshal_writesBothAHeaderAndAPayload() async throws { + let payload = Data("abcdefg".utf8) + let event = TestStream.messagewithheaderandpayload(.init( + header: "header-value", + payload: payload + )) + + let message = try await marshal(event) + + XCTAssertEqual(headers(of: message), [ + ":message-type": .string("event"), + ":event-type": .string("MessageWithHeaderAndPayload"), + "header": .string("header-value"), + ":content-type": .string("application/octet-stream"), + ]) + XCTAssertEqual(message.payload, payload) + } + + // MARK: - unbound members + + func test_marshal_writesUnboundMembersToThePayload() async throws { + let event = TestStream.messagewithnoheaderpayloadtraits(.init( + someInt: 5, + someString: "abc" + )) + + let message = try await marshal(event) + + XCTAssertEqual(headers(of: message), [ + ":message-type": .string("event"), + ":event-type": .string("MessageWithNoHeaderPayloadTraits"), + ":content-type": .string("application/json"), + ]) + XCTAssertEqual(try json(message), #"{"someInt":5,"someString":"abc"}"#) + } + + func test_marshal_writesUnboundMembersToThePayloadAndOmitsHeaderBoundMembers() async throws { + let event = TestStream.messagewithunboundpayloadtraits(.init( + header: "header-value", + unboundString: "abc" + )) + + let message = try await marshal(event) + + XCTAssertEqual(headers(of: message), [ + ":message-type": .string("event"), + ":event-type": .string("MessageWithUnboundPayloadTraits"), + "header": .string("header-value"), + ":content-type": .string("application/json"), + ]) + + // The header-bound member must not be duplicated into the payload. + XCTAssertEqual(try json(message), #"{"unboundString":"abc"}"#) + } + + func test_marshal_writesAnEmptyPayloadForAnEventWithNoMembers() async throws { + let event = TestStream.messagewithnomembers(.init()) + + let message = try await marshal(event) + + XCTAssertEqual(headers(of: message), [ + ":message-type": .string("event"), + ":event-type": .string("MessageWithNoMembers"), + ]) + XCTAssertEqual(message.payload, Data()) + } + + // MARK: - sdkUnknown + + func test_marshal_throwsForTheUnknownEventType() async throws { + let event = TestStream.sdkUnknown("unknown-event") + + do { + _ = try await marshal(event) + XCTFail("Expected an error when serializing the unknown event type") + } catch { + // expected + } + } + + // MARK: - Test helpers + + /// Serializes an input carrying a single-event stream, then decodes the request body + /// back into the message that was sent for that event. + private func marshal(_ event: TestStream) async throws -> Message { + let input = EventStreamOperationInput(eventStream: stream(of: event)) + let subject = EventStreamSerializer( + codec: SmithyAWSJSON.HTTPClientProtocol(version: .v1_0).codec, + contentType: "application/json", + messageEncoder: DefaultMessageEncoder(), + messageSigner: NoOpMessageSigner() + ) + + try input.serialize(subject) + + guard case .stream(let body) = subject.body else { + throw TestError("Expected the serialized body to be a stream") + } + let data = try await body.readToEndAsync() ?? Data() + + // Decode the encoded body back to messages. The first message is the event that + // was serialized; the empty message that terminates the stream follows it. + let decoder = DefaultMessageDecoder() + try decoder.feed(data: data) + guard let message = try decoder.message() else { + throw TestError("No message was decoded from the serialized event stream") + } + return message + } + + private func stream(of event: TestStream) -> AsyncThrowingStream { + AsyncThrowingStream { continuation in + continuation.yield(event) + continuation.finish() + } + } + + /// The message's headers, keyed by name, for order-independent comparison. + private func headers(of message: Message) -> [String: HeaderValue] { + Dictionary(uniqueKeysWithValues: message.headers.map { ($0.name, $0.value) }) + } + + private func json(_ message: Message) throws -> String { + guard let json = String(data: message.payload, encoding: .utf8) else { + throw TestError("Message payload is not valid UTF-8") + } + return json + } +} + +/// A signer that passes messages through unchanged, so tests observe only what was serialized. +private struct NoOpMessageSigner: MessageSigner { + + func sign(message: Message) async throws -> Message { message } + + func signEmpty() async throws -> Message { Message() } +} + +private struct TestError: Error { + let message: String + + init(_ message: String) { + self.message = message + } +} + diff --git a/test-sdks/build.gradle.kts b/test-sdks/build.gradle.kts index c8bf72f1b..c5f07fd15 100644 --- a/test-sdks/build.gradle.kts +++ b/test-sdks/build.gradle.kts @@ -19,6 +19,7 @@ data class TestSDK(val name: String, val forceSchemaBased: Boolean = false) val testSDKs = listOf( TestSDK("AWSJSON"), + TestSDK("EventStream"), TestSDK("JSONName"), TestSDK("HTTPLabel", true), TestSDK("HTTPQuery", true), diff --git a/test-sdks/model/eventstream.smithy b/test-sdks/model/eventstream.smithy new file mode 100644 index 000000000..37db4cefb --- /dev/null +++ b/test-sdks/model/eventstream.smithy @@ -0,0 +1,126 @@ +$version: "2.0" + +namespace smithy.swift.tests.EventStream + +use aws.protocols#awsJson1_0 + +@awsJson1_0 +service EventStream { + version: "2022-11-30" + operations: [ + EventStreamOperation + ] +} + +operation EventStreamOperation { + input: EventStreamOperationInput + output: EventStreamOperationOutput + errors: [ + EventStreamOperationError + ] +} + +structure EventStreamOperationInput { + eventStream: TestEventStream +} + +structure EventStreamOperationOutput { + eventStream: TestEventStream +} + +@error("client") +structure EventStreamOperationError { + message: String +} + +@streaming +union TestEventStream { + MessageWithBlob: MessageWithBlob + MessageWithString: MessageWithString + MessageWithStruct: MessageWithStruct + MessageWithUnion: MessageWithUnion + MessageWithHeaders: MessageWithHeaders + MessageWithHeaderAndPayload: MessageWithHeaderAndPayload + MessageWithNoHeaderPayloadTraits: MessageWithNoHeaderPayloadTraits + MessageWithUnboundPayloadTraits: MessageWithUnboundPayloadTraits + MessageWithNoMembers: MessageWithNoMembers + EventStreamOperationError: EventStreamOperationError +} + +structure MessageWithBlob { + @eventPayload + data: Blob +} + +structure MessageWithString { + @eventPayload + data: String +} + +structure MessageWithStruct { + @eventPayload + someStruct: TestStruct +} + +structure MessageWithUnion { + @eventPayload + someUnion: TestUnion +} + +structure MessageWithHeaders { + @eventHeader + blob: Blob + + @eventHeader + boolean: Boolean + + @eventHeader + byte: Byte + + @eventHeader + int: Integer + + @eventHeader + long: Long + + @eventHeader + short: Short + + @eventHeader + string: String + + @eventHeader + timestamp: Timestamp +} + +structure MessageWithHeaderAndPayload { + @eventHeader + header: String + + @eventPayload + payload: Blob +} + +structure MessageWithNoHeaderPayloadTraits { + someInt: Integer + someString: String +} + +structure MessageWithUnboundPayloadTraits { + @eventHeader + header: String + + unboundString: String +} + +structure MessageWithNoMembers {} + +structure TestStruct { + someString: String + someInt: Integer +} + +union TestUnion { + foo: String + bar: Integer +} From 01e63a0b438476f117bd7d6ba43d590b4eb7df1b Mon Sep 17 00:00:00 2001 From: Josh Elkins Date: Fri, 7 Aug 2026 08:13:56 -0500 Subject: [PATCH 2/2] Trigger CI