Implicit event stream payload members (#4535)

Author: ysaito1001

codegen-core/src/main/kotlin/software/amazon/smithy/rust/codegen/core/smithy/protocols/HttpBoundProtocolPayloadGenerator.kt

@@ -95,7 +95,12 @@ class HttpBoundProtocolPayloadGenerator(
         // need to take ownership.
         return if (payloadMemberName == null) {
             ProtocolPayloadGenerator.PayloadMetadata(takesOwnership = false)
-        } else if (operationShape.isInputEventStream(model)) {
+        } else if (
+            when (httpMessageType) {
+                HttpMessageType.REQUEST -> operationShape.isInputEventStream(model)
+                HttpMessageType.RESPONSE -> operationShape.isOutputEventStream(model)
+            }
+        ) {
             ProtocolPayloadGenerator.PayloadMetadata(takesOwnership = true)
         } else {
             val member = shape.expectMember(payloadMemberName)

codegen-core/src/main/kotlin/software/amazon/smithy/rust/codegen/core/smithy/protocols/parse/EventStreamUnmarshallerGenerator.kt

@@ -243,6 +243,26 @@ class EventStreamUnmarshallerGenerator(
                         }
                     }
                 }
+                // Handle implicit payload members: members with neither @eventPayload nor @eventHeader.
+                // Per the Smithy spec, these are collectively serialized as a protocol-specific document
+                // in the message body.
+                val implicitMembers =
+                    unionStruct.members().filter {
+                        !it.hasTrait<EventPayloadTrait>() && !it.hasTrait<EventHeaderTrait>()
+                    }
+                if (implicitMembers.isNotEmpty() && payloadMember == null) {
+                    val parser = protocol.structuredDataParser().errorParser(unionStruct)
+                    if (parser != null) {
+                        rustTemplate(
+                            """
+                            builder = #{parser}(&message.payload()[..], builder)
+                                .map_err(|err| #{Error}::unmarshalling(format!("failed to unmarshall: {err}")))?;
+                            """,
+                            "parser" to parser,
+                            *codegenScope,
+                        )
+                    }
+                }
                 rustTemplate(
                     "Ok(#{UnmarshalledMessage}::Event(#{Output}::$unionMemberName(builder.build())))",
                     "Output" to unionSymbol,