feat: support initial-request and initial-response event stream types for RPC-bound protocols (#929)

Author: lauzadis

codegen/smithy-kotlin-codegen/src/main/kotlin/software/amazon/smithy/kotlin/codegen/core/RuntimeTypes.kt

@@ -344,7 +344,12 @@ object RuntimeTypes {
         object Flow {
             // NOTE: smithy-kotlin core has an API dependency on this already
             val Flow = "kotlinx.coroutines.flow.Flow".toSymbol()
+            val flowOf = "kotlinx.coroutines.flow.flowOf".toSymbol()
+            val merge = "kotlinx.coroutines.flow.merge".toSymbol()
             val map = "kotlinx.coroutines.flow.map".toSymbol()
+            val take = "kotlinx.coroutines.flow.take".toSymbol()
+            val drop = "kotlinx.coroutines.flow.drop".toSymbol()
+            val single = "kotlinx.coroutines.flow.single".toSymbol()
         }
     }
 

codegen/smithy-kotlin-codegen/src/main/kotlin/software/amazon/smithy/kotlin/codegen/model/knowledge/SerdeIndex.kt

@@ -9,10 +9,7 @@ import software.amazon.smithy.model.Model
 import software.amazon.smithy.model.knowledge.KnowledgeIndex
 import software.amazon.smithy.model.neighbor.RelationshipType
 import software.amazon.smithy.model.neighbor.Walker
-import software.amazon.smithy.model.shapes.CollectionShape
-import software.amazon.smithy.model.shapes.OperationShape
-import software.amazon.smithy.model.shapes.Shape
-import software.amazon.smithy.model.shapes.ShapeType
+import software.amazon.smithy.model.shapes.*
 
 /**
  * Knowledge index that provides access to shapes requiring serialize and deserialize implementations.
@@ -51,11 +48,11 @@ class SerdeIndex(private val model: Model) : KnowledgeIndex {
      * Find and return the set of shapes reachable from the given shape that would require a "document" serializer.
      * @return The set of shapes that require a serializer implementation
      */
-    fun requiresDocumentSerializer(shape: Shape): Set<Shape> =
+    fun requiresDocumentSerializer(shape: Shape, members: Collection<MemberShape> = shape.members()): Set<Shape> =
         when (shape) {
             is OperationShape -> requiresDocumentSerializer(listOf(shape))
             else -> {
-                val topLevelMembers = shape.members()
+                val topLevelMembers = members
                     .map { model.expectShape(it.target) }
                     .filter { it.isStructureShape || it.isUnionShape || it is CollectionShape || it.isMapShape }
                     .toSet()
@@ -104,11 +101,11 @@ class SerdeIndex(private val model: Model) : KnowledgeIndex {
      * Find and return the set of shapes reachable from the given shape that would require a "document" deserializer.
      * @return The set of shapes that require a deserializer implementation
      */
-    fun requiresDocumentDeserializer(shape: Shape): Set<Shape> =
+    fun requiresDocumentDeserializer(shape: Shape, members: Collection<MemberShape> = shape.members()): Set<Shape> =
         when (shape) {
             is OperationShape -> requiresDocumentDeserializer(listOf(shape))
             else -> {
-                val topLevelMembers = shape.members()
+                val topLevelMembers = members
                     .map { model.expectShape(it.target) }
                     .filter { it.isStructureShape || it.isUnionShape || it is CollectionShape || it.isMapShape }
                     .toMutableSet()

codegen/smithy-kotlin-codegen/src/main/kotlin/software/amazon/smithy/kotlin/codegen/rendering/serde/JsonParserGenerator.kt

@@ -40,10 +40,14 @@ open class JsonParserGenerator(
     /**
      * Register nested structure/map shapes reachable from the operation input shape that require a "document" deserializer
      * implementation
+     * @param ctx the generation context
+     * @param shape the shape to generated nested document deserializers for
+     * @param writer the writer to write with
+     * @param members the subset of shapes to generated nested document deserializers for
      */
-    private fun addNestedDocumentDeserializers(ctx: ProtocolGenerator.GenerationContext, shape: Shape, writer: KotlinWriter) {
+    private fun addNestedDocumentDeserializers(ctx: ProtocolGenerator.GenerationContext, shape: Shape, writer: KotlinWriter, members: Collection<MemberShape> = shape.members()) {
         val serdeIndex = SerdeIndex.of(ctx.model)
-        val shapesRequiringDocumentDeserializer = serdeIndex.requiresDocumentDeserializer(shape)
+        val shapesRequiringDocumentDeserializer = serdeIndex.requiresDocumentDeserializer(shape, members)
 
         // register a dependency on each of the members that require a deserializer impl
         // ensuring they get generated
@@ -141,7 +145,7 @@ open class JsonParserGenerator(
         val forMembers = members ?: target.members()
         val deserializeFn = documentDeserializer(ctx, target, forMembers)
         return target.payloadDeserializer(ctx.settings, symbol, forMembers) { writer ->
-            addNestedDocumentDeserializers(ctx, target, writer)
+            addNestedDocumentDeserializers(ctx, target, writer, forMembers)
             writer.withBlock("internal fun #identifier.name:L(payload: ByteArray): #T {", "}", symbol) {
                 if (target.members().isEmpty() && !target.isDocumentShape) {
                     // short circuit when the shape has no modeled members to deserialize

codegen/smithy-kotlin-codegen/src/main/kotlin/software/amazon/smithy/kotlin/codegen/rendering/serde/JsonSerializerGenerator.kt

@@ -44,9 +44,9 @@ open class JsonSerializerGenerator(
      * Register nested structure/map shapes reachable from the operation input shape that require a "document" serializer
      * implementation
      */
-    private fun addNestedDocumentSerializers(ctx: ProtocolGenerator.GenerationContext, shape: Shape, writer: KotlinWriter) {
+    private fun addNestedDocumentSerializers(ctx: ProtocolGenerator.GenerationContext, shape: Shape, writer: KotlinWriter, members: Collection<MemberShape> = shape.members()) {
         val serdeIndex = SerdeIndex.of(ctx.model)
-        val shapesRequiringDocumentSerializer = serdeIndex.requiresDocumentSerializer(shape)
+        val shapesRequiringDocumentSerializer = serdeIndex.requiresDocumentSerializer(shape, members)
         // register a dependency on each of the members that require a serializer impl
         // ensuring they get generated
         shapesRequiringDocumentSerializer.forEach {
@@ -109,7 +109,7 @@ open class JsonSerializerGenerator(
 
         val serializeFn = documentSerializer(ctx, target, forMembers)
         return target.payloadSerializer(ctx.settings, symbol, forMembers) { writer ->
-            addNestedDocumentSerializers(ctx, target, writer)
+            addNestedDocumentSerializers(ctx, target, writer, forMembers)
             writer.addImportReferences(symbol, SymbolReference.ContextOption.USE)
             writer.withBlock("internal fun #identifier.name:L(input: #T): ByteArray {", "}", symbol) {
                 write("val serializer = #T()", RuntimeTypes.Serde.SerdeJson.JsonSerializer)