Add event streaming signing support

auth-api/src/main/java/software/amazon/smithy/java/auth/api/NullSigner.java

@@ -30,7 +30,7 @@ private NullSigner() {}
      * @return the request as-is.
      */
     @Override
-    public Object sign(Object request, Identity identity, Context properties) {
-        return request;
+    public SignResult<Object> sign(Object request, Identity identity, Context properties) {
+        return new SignResult<>(request);
     }
 }

auth-api/src/main/java/software/amazon/smithy/java/auth/api/SignResult.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package software.amazon.smithy.java.auth.api;
+
+/**
+ * Holds the result of signing a request of type {@link RequestT}.
+ *
+ * @param signedRequest the signed request
+ * @param signature     the signature
+ * @param <RequestT>    the type of the request
+ */
+public record SignResult<RequestT>(
+        RequestT signedRequest,
+        String signature) {
+
+    /**
+     * Creates a sign result with an empty string
+     *
+     * @param signedRequest the signed request
+     */
+    public SignResult(RequestT signedRequest) {
+        this(signedRequest, "");
+    }
+}

auth-api/src/main/java/software/amazon/smithy/java/auth/api/Signer.java

@@ -24,7 +24,7 @@ public interface Signer<RequestT, IdentityT extends Identity> extends AutoClosea
      * @param properties Signing properties.
      * @return the signed request.
      */
-    RequestT sign(RequestT request, IdentityT identity, Context properties);
+    SignResult<RequestT> sign(RequestT request, IdentityT identity, Context properties);
 
     @SuppressWarnings("unchecked")
     static <RequestT, IdentityT extends Identity> Signer<RequestT, IdentityT> nullSigner() {

aws/aws-event-streams/src/main/java/software/amazon/smithy/java/aws/events/AwsEventDecoderFactory.java

@@ -16,7 +16,7 @@
 import software.amazon.smithy.java.core.serde.event.EventDecoder;
 import software.amazon.smithy.java.core.serde.event.EventDecoderFactory;
 import software.amazon.smithy.java.core.serde.event.FrameDecoder;
-import software.amazon.smithy.java.core.serde.event.FrameTransformer;
+import software.amazon.smithy.java.core.serde.event.FrameProcessor;
 
 /**
  * A {@link EventDecoderFactory} for AWS events.
@@ -32,69 +32,69 @@ public final class AwsEventDecoderFactory<E extends SerializableStruct, IR exten
     private final Schema eventSchema;
     private final Codec codec;
     private final Supplier<ShapeBuilder<E>> eventBuilder;
-    private final FrameTransformer<AwsEventFrame> transformer;
+    private final FrameProcessor<AwsEventFrame> frameProcessor;
 
     private AwsEventDecoderFactory(
             InitialEventType initialEventType,
             Supplier<ShapeBuilder<IR>> initialEventBuilder,
             Schema eventSchema,
             Codec codec,
             Supplier<ShapeBuilder<E>> eventBuilder,
-            FrameTransformer<AwsEventFrame> transformer
+            FrameProcessor<AwsEventFrame> frameProcessor
     ) {
         this.initialEventType = Objects.requireNonNull(initialEventType, "initialEventType");
         this.initialEventBuilder = Objects.requireNonNull(initialEventBuilder, "initialEventBuilder");
         this.eventSchema = Objects.requireNonNull(eventSchema, "eventSchema").isMember() ? eventSchema.memberTarget()
                 : eventSchema;
         this.codec = Objects.requireNonNull(codec, "codec");
         this.eventBuilder = Objects.requireNonNull(eventBuilder, "eventBuilder");
-        this.transformer = Objects.requireNonNull(transformer, "transformer");
+        this.frameProcessor = Objects.requireNonNull(frameProcessor, "transformer");
     }
 
     /**
      * Creates a new input stream decoder factory.
      *
-     * @param operation   The input operation for the factory
-     * @param codec       The protocol codec to decode the payload
-     * @param transformer The frame transformer
-     * @param <IE>        The output event type
+     * @param operation      The input operation for the factory
+     * @param codec          The protocol codec to decode the payload
+     * @param frameProcessor The frame frameProcessor
+     * @param <IE>           The output event type
      * @return A new event decoder factory
      */
     public static <IE extends SerializableStruct> AwsEventDecoderFactory<IE, ?> forInputStream(
             InputEventStreamingApiOperation<?, ?, IE> operation,
             Codec codec,
-            FrameTransformer<AwsEventFrame> transformer
+            FrameProcessor<AwsEventFrame> frameProcessor
     ) {
         return new AwsEventDecoderFactory<>(
                 InitialEventType.INITIAL_REQUEST,
                 operation::inputBuilder,
                 operation.inputStreamMember(),
                 codec,
                 operation.inputEventBuilderSupplier(),
-                transformer);
+                frameProcessor);
     }
 
     /**
      * Creates a new output stream decoder factory.
      *
-     * @param operation   The output operation for the factory
-     * @param codec       The protocol codec to decode the payload
-     * @param transformer The frame transformer
-     * @param <OE>        The output event type
+     * @param operation      The output operation for the factory
+     * @param codec          The protocol codec to decode the payload
+     * @param frameProcessor The frame frameProcessor
+     * @param <OE>           The output event type
      * @return A new event decoder factory
      */
     public static <OE extends SerializableStruct> AwsEventDecoderFactory<OE, ?> forOutputStream(
             OutputEventStreamingApiOperation<?, ?, OE> operation,
             Codec codec,
-            FrameTransformer<AwsEventFrame> transformer
+            FrameProcessor<AwsEventFrame> frameProcessor
     ) {
         return new AwsEventDecoderFactory<>(
                 InitialEventType.INITIAL_RESPONSE,
                 operation::outputBuilder,
                 operation.outputStreamMember(),
                 codec,
                 operation.outputEventBuilderSupplier(),
-                transformer);
+                frameProcessor);
     }
 
     @Override
@@ -104,6 +104,6 @@ public EventDecoder<AwsEventFrame> newEventDecoder() {
 
     @Override
     public FrameDecoder<AwsEventFrame> newFrameDecoder() {
-        return new AwsFrameDecoder(transformer);
+        return new AwsFrameDecoder(frameProcessor);
     }
 }

aws/aws-event-streams/src/main/java/software/amazon/smithy/java/aws/events/AwsEventEncoderFactory.java

@@ -15,7 +15,7 @@
 import software.amazon.smithy.java.core.serde.event.EventEncoderFactory;
 import software.amazon.smithy.java.core.serde.event.EventStreamingException;
 import software.amazon.smithy.java.core.serde.event.FrameEncoder;
-import software.amazon.smithy.java.core.serde.event.FrameTransformer;
+import software.amazon.smithy.java.core.serde.event.FrameProcessor;
 
 /**
  * A {@link EventEncoderFactory} for AWS events.
@@ -25,22 +25,22 @@ public final class AwsEventEncoderFactory implements EventEncoderFactory<AwsEven
     private final Schema schema;
     private final Codec codec;
     private final String payloadMediaType;
-    private final FrameTransformer<AwsEventFrame> transformer;
+    private final FrameProcessor<AwsEventFrame> frameProcessor;
     private final Function<Throwable, EventStreamingException> exceptionHandler;
 
     private AwsEventEncoderFactory(
             InitialEventType initialEventType,
             Schema schema,
             Codec codec,
             String payloadMediaType,
-            FrameTransformer<AwsEventFrame> transformer,
+            FrameProcessor<AwsEventFrame> frameProcessor,
             Function<Throwable, EventStreamingException> exceptionHandler
     ) {
         this.initialEventType = Objects.requireNonNull(initialEventType, "initialEventType");
         this.schema = Objects.requireNonNull(schema, "schema").isMember() ? schema.memberTarget() : schema;
         this.codec = Objects.requireNonNull(codec, "codec");
         this.payloadMediaType = Objects.requireNonNull(payloadMediaType, "payloadMediaType");
-        this.transformer = Objects.requireNonNull(transformer, "transformer");
+        this.frameProcessor = Objects.requireNonNull(frameProcessor, "frameProcessor");
         this.exceptionHandler = Objects.requireNonNull(exceptionHandler, "exceptionHandler");
     }
 
@@ -57,7 +57,7 @@ public static AwsEventEncoderFactory forInputStream(
             InputEventStreamingApiOperation<?, ?, ?> operation,
             Codec codec,
             String payloadMediaType,
-            FrameTransformer<AwsEventFrame> transformer,
+            FrameProcessor<AwsEventFrame> transformer,
             Function<Throwable, EventStreamingException> exceptionHandler
     ) {
         return new AwsEventEncoderFactory(InitialEventType.INITIAL_REQUEST,
@@ -81,7 +81,7 @@ public static AwsEventEncoderFactory forOutputStream(
             OutputEventStreamingApiOperation<?, ?, ?> operation,
             Codec codec,
             String payloadMediaType,
-            FrameTransformer<AwsEventFrame> transformer,
+            FrameProcessor<AwsEventFrame> transformer,
             Function<Throwable, EventStreamingException> exceptionHandler
     ) {
         return new AwsEventEncoderFactory(InitialEventType.INITIAL_RESPONSE,
@@ -98,7 +98,7 @@ public EventEncoder<AwsEventFrame> newEventEncoder() {
                 schema,
                 codec,
                 payloadMediaType,
-                transformer,
+                frameProcessor,
                 exceptionHandler);
     }
 
@@ -111,4 +111,14 @@ public FrameEncoder<AwsEventFrame> newFrameEncoder() {
     public String contentType() {
         return "application/vnd.amazon.eventstream";
     }
+
+    @Override
+    public EventEncoderFactory<AwsEventFrame> withFrameProcessor(FrameProcessor<AwsEventFrame> frameProcessor) {
+        return new AwsEventEncoderFactory(initialEventType,
+                schema,
+                codec,
+                payloadMediaType,
+                frameProcessor,
+                exceptionHandler);
+    }
 }

aws/aws-event-streams/src/main/java/software/amazon/smithy/java/aws/events/AwsEventFrame.java

@@ -12,7 +12,7 @@ public final class AwsEventFrame implements Frame<Message> {
 
     private final Message message;
 
-    AwsEventFrame(Message message) {
+    public AwsEventFrame(Message message) {
         this.message = message;
     }
 

aws/aws-event-streams/src/main/java/software/amazon/smithy/java/aws/events/AwsEventShapeDecoder.java

@@ -58,6 +58,7 @@ public SerializableStruct decode(AwsEventFrame frame) {
 
     private E decodeEvent(AwsEventFrame frame) {
         var message = frame.unwrap();
+        // TODO Add support for :message-type other than "event".
         var eventType = getEventType(message);
         var memberSchema = eventSchema.member(eventType);
         if (memberSchema == null) {
@@ -99,6 +100,10 @@ private Schema getEventStreamMember(Schema schema) {
         throw new IllegalArgumentException("cannot find streaming member");
     }
 
+    private String getMessageType(Message message) {
+        return message.getHeaders().get(":message-type").getString();
+    }
+
     private String getEventType(Message message) {
         return message.getHeaders().get(":event-type").getString();
     }

aws/aws-event-streams/src/main/java/software/amazon/smithy/java/aws/events/AwsEventShapeEncoder.java

@@ -26,7 +26,7 @@
 import software.amazon.smithy.java.core.serde.SpecificShapeSerializer;
 import software.amazon.smithy.java.core.serde.event.EventEncoder;
 import software.amazon.smithy.java.core.serde.event.EventStreamingException;
-import software.amazon.smithy.java.core.serde.event.FrameTransformer;
+import software.amazon.smithy.java.core.serde.event.FrameProcessor;
 import software.amazon.smithy.model.shapes.ShapeId;
 
 public final class AwsEventShapeEncoder implements EventEncoder<AwsEventFrame> {
@@ -36,15 +36,15 @@ public final class AwsEventShapeEncoder implements EventEncoder<AwsEventFrame> {
     private final String payloadMediaType;
     private final Map<String, BiFunction<OutputStream, Map<String, HeaderValue>, ShapeSerializer>> possibleTypes;
     private final Map<ShapeId, Schema> possibleExceptions;
-    private final FrameTransformer<AwsEventFrame> frameTransformer;
+    private final FrameProcessor<AwsEventFrame> frameProcessor;
     private final Function<Throwable, EventStreamingException> exceptionHandler;
 
     public AwsEventShapeEncoder(
             InitialEventType initialEventType,
             Schema eventSchema,
             Codec codec,
             String payloadMediaType,
-            FrameTransformer<AwsEventFrame> frameTransformer,
+            FrameProcessor<AwsEventFrame> frameProcessor,
             Function<Throwable, EventStreamingException> exceptionHandler
     ) {
         this.initialEventType = Objects.requireNonNull(initialEventType, "initialEventType");
@@ -54,7 +54,7 @@ public AwsEventShapeEncoder(
                 codec,
                 initialEventType.value());
         this.possibleExceptions = possibleExceptions(Objects.requireNonNull(eventSchema, "eventSchema"));
-        this.frameTransformer = Objects.requireNonNull(frameTransformer, "frameTransformer");
+        this.frameProcessor = Objects.requireNonNull(frameProcessor, "frameTransformer");
         this.exceptionHandler = Objects.requireNonNull(exceptionHandler, "exceptionHandler");
     }
 
@@ -67,7 +67,7 @@ public AwsEventFrame encode(SerializableStruct item) {
         headers.put(":event-type", HeaderValue.fromString(typeHolder.get()));
         headers.put(":content-type", HeaderValue.fromString(payloadMediaType));
         var frame = new AwsEventFrame(new Message(headers, payload));
-        return frameTransformer.apply(frame);
+        return frameProcessor.transformFrame(frame);
     }
 
     private byte[] encodeInput(
@@ -154,6 +154,11 @@ public AwsEventFrame encodeFailure(Throwable exception) {
 
     }
 
+    @Override
+    public AwsEventFrame closingFrame() {
+        return frameProcessor.closingFrame();
+    }
+
     static Map<String, BiFunction<OutputStream, Map<String, HeaderValue>, ShapeSerializer>> possibleTypes(
             Schema eventSchema,
             Codec codec,

aws/aws-event-streams/src/main/java/software/amazon/smithy/java/aws/events/AwsFrameDecoder.java

@@ -10,14 +10,14 @@
 import java.util.List;
 import software.amazon.eventstream.MessageDecoder;
 import software.amazon.smithy.java.core.serde.event.FrameDecoder;
-import software.amazon.smithy.java.core.serde.event.FrameTransformer;
+import software.amazon.smithy.java.core.serde.event.FrameProcessor;
 
 public final class AwsFrameDecoder implements FrameDecoder<AwsEventFrame> {
     private final MessageDecoder decoder = new MessageDecoder();
-    private final FrameTransformer<AwsEventFrame> transformer;
+    private final FrameProcessor<AwsEventFrame> frameProcessor;
 
-    public AwsFrameDecoder(FrameTransformer<AwsEventFrame> transformer) {
-        this.transformer = transformer;
+    public AwsFrameDecoder(FrameProcessor<AwsEventFrame> frameProcessor) {
+        this.frameProcessor = frameProcessor;
     }
 
     @Override
@@ -27,7 +27,7 @@ public List<AwsEventFrame> decode(ByteBuffer buffer) {
         var result = new ArrayList<AwsEventFrame>(messages.size());
         for (var message : messages) {
             var event = new AwsEventFrame(message);
-            var transformed = transformer.apply(event);
+            var transformed = frameProcessor.transformFrame(event);
             if (transformed != null) {
                 result.add(transformed);
             }

aws/aws-event-streams/src/test/java/software/amazon/smithy/java/aws/events/AwsEventShapeEncoderTest.java

@@ -20,7 +20,7 @@
 import software.amazon.smithy.java.aws.events.model.TestOperationInput;
 import software.amazon.smithy.java.core.serde.Codec;
 import software.amazon.smithy.java.core.serde.event.EventStreamingException;
-import software.amazon.smithy.java.core.serde.event.FrameTransformer;
+import software.amazon.smithy.java.core.serde.event.FrameProcessor;
 import software.amazon.smithy.java.json.JsonCodec;
 
 class AwsEventShapeEncoderTest {
@@ -136,7 +136,7 @@ static AwsEventShapeEncoder createEncoder() {
                 TestOperation.instance().inputStreamMember(), // event schema
                 createJsonCodec(), // codec
                 "text/json",
-                FrameTransformer.identity(),
+                FrameProcessor.identity(),
                 (e) -> new EventStreamingException("InternalServerException", "Internal Server Error"));
     }
 

aws/aws-sigv4/build.gradle.kts

@@ -14,6 +14,7 @@ extra["moduleName"] = "software.amazon.smithy.java.aws.sigv4"
 dependencies {
     implementation(project(":client:client-core"))
     api(project(":aws:client:aws-client-core"))
+    api(project(":aws:aws-event-streams"))
     implementation(project(":http:http-api"))
     implementation(project(":io"))
     implementation(project(":logging"))

aws/aws-sigv4/src/main/java/software/amazon/smithy/java/aws/client/auth/scheme/sigv4/SigV4AuthScheme.java

@@ -8,9 +8,11 @@
 import software.amazon.smithy.aws.traits.auth.SigV4Trait;
 import software.amazon.smithy.java.auth.api.Signer;
 import software.amazon.smithy.java.aws.auth.api.identity.AwsCredentialsIdentity;
+import software.amazon.smithy.java.aws.events.AwsEventFrame;
 import software.amazon.smithy.java.client.core.auth.scheme.AuthScheme;
 import software.amazon.smithy.java.client.core.auth.scheme.AuthSchemeFactory;
 import software.amazon.smithy.java.context.Context;
+import software.amazon.smithy.java.core.serde.event.FrameProcessor;
 import software.amazon.smithy.java.http.api.HttpRequest;
 import software.amazon.smithy.model.shapes.ShapeId;
 
@@ -72,6 +74,16 @@ public Signer<HttpRequest, AwsCredentialsIdentity> signer() {
         return SigV4Signer.create();
     }
 
+    @Override
+    @SuppressWarnings("unchecked")
+    public FrameProcessor<AwsEventFrame> eventSigner(
+            AwsCredentialsIdentity identity,
+            Context context,
+            String seedSignature
+    ) {
+        return new SigV4EventSigner(identity, context, seedSignature);
+    }
+
     public static final class Factory implements AuthSchemeFactory<SigV4Trait> {
 
         @Override