Add support for event stream requests over RPC

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"
+    }
+  }
+}