Add support for event stream requests over RPC

This commit adds support for RPC services that use client->server or
bidirectional event streaming. In the RPC formats (i.e. AWS JSON), the
marshalled request and response objects are sent as the first event in
the request and response stream respectively. This is different from the
REST formats, such as REST JSON, which bind the request and response
fields to non-payload positions (headers, URI, method, status code,
query string) when event streaming is used.

The implementation of this feature is mainly performed by a new global
interceptor, `EventStreamInitialRequestInterceptor`. The interceptor
detects when an `initial-request` event needs to be sent and transforms
the async response body by constructing the `initial-request` message
and prepending it to the event stream. Due to the high precedence of
global interceptors, any subsequent service or override interceptors
that run on the body will see the transformed version. This also applies
to the `EventStreamAws4Signer`, which will produce a signature over the
`initial-request` event.

Author: rschmitt

codegen/src/main/java/software/amazon/awssdk/codegen/poet/client/specs/JsonProtocolSpec.java

@@ -232,6 +232,7 @@ public CodeBlock asyncExecutionHandler(IntermediateModel intermediateModel, Oper
                .add(".withMarshaller($L)\n", asyncMarshaller(model, opModel, marshaller, protocolFactory))
                .add(asyncRequestBody(opModel))
                .add(fullDuplex(opModel))
+               .add(hasInitialRequestEvent(opModel, isRestJson))
                .add(".withResponseHandler($L)\n", responseHandlerName(opModel, isRestJson))
                .add(".withErrorResponseHandler(errorResponseHandler)\n")
                .add(".withMetricCollector(apiCallMetricCollector)\n")
@@ -271,6 +272,11 @@ private CodeBlock fullDuplex(OperationModel opModel) {
                                                                                : CodeBlock.of("");
     }
 
+    private CodeBlock hasInitialRequestEvent(OperationModel opModel, boolean isRestJson) {
+        return opModel.hasEventStreamInput() && !isRestJson ? CodeBlock.of(".withInitialRequestEvent(true)")
+                                                            : CodeBlock.of("");
+    }
+
     private CodeBlock asyncRequestBody(OperationModel opModel) {
         return opModel.hasEventStreamInput() ? CodeBlock.of(".withAsyncRequestBody($T.fromPublisher(adapted))",
                                                             AsyncRequestBody.class)

codegen/src/main/java/software/amazon/awssdk/codegen/poet/transform/protocols/EventStreamJsonMarshallerSpec.java

@@ -32,8 +32,6 @@
  */
 public final class EventStreamJsonMarshallerSpec extends JsonMarshallerSpec {
 
-    private static final String JSON_CONTENT_TYPE = "application/json";
-
     public EventStreamJsonMarshallerSpec(IntermediateModel model, ShapeModel shapeModel) {
         super(shapeModel);
     }
@@ -51,7 +49,7 @@ public CodeBlock marshalCodeBlock(ClassName requestClassName) {
 
         // Add :content-type header only if payload is present
         if (!shapeModel.hasNoEventPayload()) {
-            builder.add(".putHeader(\":content-type\", \"$L\")", determinePayloadContentType());
+            builder.add(".putHeader(\":content-type\", $L)", determinePayloadContentType());
         }
 
         builder.add(".build();");
@@ -85,12 +83,12 @@ private String determinePayloadContentType() {
             return getPayloadContentType(explicitEventPayload);
         }
 
-        return JSON_CONTENT_TYPE;
+        return "protocolFactory.getContentType()";
     }
 
     private String getPayloadContentType(MemberModel memberModel) {
-        String blobContentType = "application/octet-stream";
-        String stringContentType = "text/plain";
+        String blobContentType = "\"application/octet-stream\"";
+        String stringContentType = "\"text/plain\"";
         String variableType = memberModel.getVariable().getVariableType();
 
         if ("software.amazon.awssdk.core.SdkBytes".equals(variableType)) {
@@ -99,6 +97,6 @@ private String getPayloadContentType(MemberModel memberModel) {
             return stringContentType;
         }
 
-        return JSON_CONTENT_TYPE;
+        return "protocolFactory.getContentType()";
     }
 }

codegen/src/test/java/software/amazon/awssdk/codegen/poet/ClientTestModels.java

@@ -16,6 +16,8 @@
 package software.amazon.awssdk.codegen.poet;
 
 import java.io.File;
+
+import org.eclipse.core.runtime.internal.adaptor.IModel;
 import software.amazon.awssdk.codegen.C2jModels;
 import software.amazon.awssdk.codegen.IntermediateModelBuilder;
 import software.amazon.awssdk.codegen.model.config.customization.CustomizationConfig;
@@ -31,7 +33,20 @@
 public class ClientTestModels {
     private ClientTestModels() {}
 
-    public static IntermediateModel jsonServiceModels() {
+    public static IntermediateModel awsJsonServiceModels() {
+        File serviceModel = new File(ClientTestModels.class.getResource("client/c2j/json/service-2.json").getFile());
+        File customizationModel = new File(ClientTestModels.class.getResource("client/c2j/json/customization.config").getFile());
+        File paginatorsModel = new File(ClientTestModels.class.getResource("client/c2j/json/paginators.json").getFile());
+        C2jModels models = C2jModels.builder()
+            .serviceModel(getServiceModel(serviceModel))
+            .customizationConfig(getCustomizationConfig(customizationModel))
+            .paginatorsModel(getPaginatorsModel(paginatorsModel))
+            .build();
+
+        return new IntermediateModelBuilder(models).build();
+    }
+
+    public static IntermediateModel restJsonServiceModels() {
         File serviceModel = new File(ClientTestModels.class.getResource("client/c2j/rest-json/service-2.json").getFile());
         File customizationModel = new File(ClientTestModels.class.getResource("client/c2j/rest-json/customization.config").getFile());
         File paginatorsModel = new File(ClientTestModels.class.getResource("client/c2j/rest-json/paginators.json").getFile());

codegen/src/test/java/software/amazon/awssdk/codegen/poet/builder/BuilderClassTest.java

@@ -69,7 +69,7 @@ public void asyncClientBuilderClass() throws Exception {
     }
 
     private void validateGeneration(Function<IntermediateModel, ClassSpec> generatorConstructor, String expectedClassName) {
-        assertThat(generatorConstructor.apply(ClientTestModels.jsonServiceModels()), generatesTo(expectedClassName));
+        assertThat(generatorConstructor.apply(ClientTestModels.restJsonServiceModels()), generatesTo(expectedClassName));
     }
 
     private void validateQueryGeneration(Function<IntermediateModel, ClassSpec> generatorConstructor, String expectedClassName) {

codegen/src/test/java/software/amazon/awssdk/codegen/poet/client/PoetClientFunctionalTests.java

@@ -28,27 +28,26 @@ public class PoetClientFunctionalTests {
 
     @Test
     public void asyncClientClass() throws Exception {
-        AsyncClientClass asyncClientClass = new AsyncClientClass(
-                GeneratorTaskParams.create(ClientTestModels.jsonServiceModels(), "sources/", "tests/"));
+        AsyncClientClass asyncClientClass = createAsyncClientClass(ClientTestModels.restJsonServiceModels());
         assertThat(asyncClientClass, generatesTo("test-async-client-class.java"));
     }
 
     @Test
     public void asyncClientInterface() throws Exception {
-        ClassSpec asyncClientInterface = new AsyncClientInterface(ClientTestModels.jsonServiceModels());
+        ClassSpec asyncClientInterface = new AsyncClientInterface(ClientTestModels.restJsonServiceModels());
         assertThat(asyncClientInterface, generatesTo("test-json-async-client-interface.java"));
     }
 
     @Test
     public void simpleMethodsIntegClass() throws Exception {
         ClientSimpleMethodsIntegrationTests simpleMethodsClass = new ClientSimpleMethodsIntegrationTests(
-                ClientTestModels.jsonServiceModels());
+                ClientTestModels.restJsonServiceModels());
         assertThat(simpleMethodsClass, generatesTo("test-simple-methods-integ-class.java"));
     }
 
     @Test
-    public void syncClientClassJson() throws Exception {
-        SyncClientClass syncClientClass = createSyncClientClass(ClientTestModels.jsonServiceModels());
+    public void syncClientClassRestJson() throws Exception {
+        SyncClientClass syncClientClass = createSyncClientClass(ClientTestModels.restJsonServiceModels());
         assertThat(syncClientClass, generatesTo("test-json-client-class.java"));
     }
 
@@ -58,6 +57,11 @@ public void syncClientClassQuery() throws Exception {
         assertThat(syncClientClass, generatesTo("test-query-client-class.java"));
     }
 
+    @Test
+    public void asyncClientClassAwsJson() throws Exception {
+        AsyncClientClass asyncClientClass = createAsyncClientClass(ClientTestModels.awsJsonServiceModels());
+        assertThat(asyncClientClass, generatesTo("test-aws-json-async-client-class.java"));
+    }
 
     @Test
     public void asyncClientClassQuery() throws Exception {
@@ -71,7 +75,6 @@ public void syncClientClassXml() throws Exception {
         assertThat(syncClientClass, generatesTo("test-xml-client-class.java"));
     }
 
-
     @Test
     public void asyncClientClassXml() throws Exception {
         AsyncClientClass syncClientClass = createAsyncClientClass(ClientTestModels.xmlServiceModels());
@@ -88,7 +91,7 @@ private AsyncClientClass createAsyncClientClass(IntermediateModel model) {
 
     @Test
     public void syncClientInterface() throws Exception {
-        ClassSpec syncClientInterface = new SyncClientInterface(ClientTestModels.jsonServiceModels());
+        ClassSpec syncClientInterface = new SyncClientInterface(ClientTestModels.restJsonServiceModels());
         assertThat(syncClientInterface, generatesTo("test-json-client-interface.java"));
     }
 

codegen/src/test/java/software/amazon/awssdk/codegen/poet/eventstream/EventStreamFunctionalTests.java

@@ -45,7 +45,7 @@ public void responseHandlerBuilder() throws Exception {
 
     private void runTest(BiFunction<GeneratorTaskParams, OperationModel, ClassSpec> specFactory,
                          String expectedTestFile) {
-        IntermediateModel model = ClientTestModels.jsonServiceModels();
+        IntermediateModel model = ClientTestModels.restJsonServiceModels();
         GeneratorTaskParams dependencies = GeneratorTaskParams.create(model, "sources/", "tests/");
         ClassSpec classSpec = specFactory.apply(dependencies, model.getOperation("EventStreamOperation"));
         assertThat(classSpec, generatesTo(expectedTestFile));

codegen/src/test/resources/software/amazon/awssdk/codegen/poet/client/c2j/json/customization.config

@@ -0,0 +1,20 @@
+{
+    "authPolicyActions" : {
+        "skip" : true
+    },
+    "presignersFqcn": "software.amazon.awssdk.services.acm.presign.AcmClientPresigners",
+    "serviceSpecificHttpConfig": "software.amazon.MyServiceHttpConfig",
+    "serviceSpecificClientConfigClass": "ServiceConfiguration",
+    "customRetryPolicy": "software.amazon.MyServiceRetryPolicy",
+    "verifiedSimpleMethods" : ["paginatedOperationWithResultKey"],
+    "blacklistedSimpleMethods" : [
+        "eventStreamOperation"
+    ],
+  "utilitiesMethod": {
+    "returnType": "software.amazon.awssdk.services.json.JsonUtilities",
+    "createMethodParams": ["param1", "param2", "param3"]
+  },
+  "useLegacyEventGenerationScheme": {
+    "EventStream": ["EventOne", "event-two", "eventThree"]
+  }
+}

codegen/src/test/resources/software/amazon/awssdk/codegen/poet/client/c2j/json/paginators.json

@@ -0,0 +1,15 @@
+{
+  "pagination": {
+    "PaginatedOperationWithResultKey": {
+      "input_token": "NextToken",
+      "output_token": "NextToken",
+      "limit_key": "MaxResults",
+      "result_key": "Items"
+    },
+    "PaginatedOperationWithoutResultKey": {
+      "input_token": "NextToken",
+      "output_token": "NextToken",
+      "limit_key": "MaxResults"
+    }
+  }
+}