Remove futures_core::stream::Stream from `aws_smithy_http::byte_str…

…eam::ByteStream` (#2983)

## Motivation and Context
Removes `futures_core::stream::Stream` from
`aws_smithy_http::byte_stream::ByteStream`

## Description
This PR is part of our ongoing effort,
#2413. We remove the
`futures_core::stream::Stream` trait from
`aws_smithy_http::byte_stream::ByteStream`. To continue support existing
places that relied upon `ByteStream` implementing the `Stream` trait, we
provide explicit `.next`, `.try_next`, and `.size_hints` methods on that
type. As part of it, a doc-hidden type `FuturesStreamCompatByteStream`,
which implements `Stream`, has been added to `aws_smithy_http` to cater
to those who need a type implementing `Stream` (see
[discussion](#2910 (comment))
why doc-hidden).

Another place we need to update is codegen responsible for rendering
stream payload serializer, and this affects the server. The regular
server and the python server have slightly different rendering
requirements, since the former uses
`aws_smithy_http::byte_stream::ByteStream` (that does _not_ implement
the `Stream` trait) and latter uses its own
[ByteStream](https://github.com/awslabs/smithy-rs/blob/cb79a68e3c38d1e62d3980d5e7baedc1144bacc7/rust-runtime/aws-smithy-http-server-python/src/types.rs#L343)
(that does implement `Stream`). We use
`ServerHttpBoundProtocolCustomization` to handle the difference:
`StreamPayloadSerializerCustomization` and
`PythonServerStreamPayloadSerializerCustomization`.

## Testing
No new behavior has been added, relied on the existing tests in CI.

## Checklist
- [x] I have updated `CHANGELOG.next.toml` if I made changes to the
smithy-rs codegen or runtime crates
- [x] I have updated `CHANGELOG.next.toml` if I made changes to the AWS
SDK, generated SDK code, or SDK runtime crates

----

_By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice._

CHANGELOG.next.toml

@@ -229,3 +229,15 @@ message = "Add support for Sigv4A request signing. Sigv4a signing will be used a
 references = ["smithy-rs#1797"]
 meta = { "breaking" = true, "tada" = true, "bug" = false }
 author = "Velfi"
+
+[[aws-sdk-rust]]
+message = "The `futures_core::stream::Stream` trait has been removed from [`ByteStream`](https://docs.rs/aws-smithy-http/latest/aws_smithy_http/byte_stream/struct.ByteStream.html). The methods mentioned in the [doc](https://docs.rs/aws-smithy-http/latest/aws_smithy_http/byte_stream/struct.ByteStream.html#getting-data-out-of-a-bytestream) will continue to be supported. Other stream operations that were previously available through the trait or its extension traits can be added later in a backward compatible manner."
+references = ["smithy-rs#2983"]
+meta = { "breaking" = true, "tada" = false, "bug" = false }
+author = "ysaito1001"
+
+[[smithy-rs]]
+message = "The `futures_core::stream::Stream` trait has been removed from [`ByteStream`](https://docs.rs/aws-smithy-http/latest/aws_smithy_http/byte_stream/struct.ByteStream.html). The methods mentioned in the [doc](https://docs.rs/aws-smithy-http/latest/aws_smithy_http/byte_stream/struct.ByteStream.html#getting-data-out-of-a-bytestream) will continue to be supported. Other stream operations that were previously available through the trait or its extension traits can be added later in a backward compatible manner."
+references = ["smithy-rs#2983"]
+meta = { "breaking" = true, "tada" = false, "bug" = false, "target" = "client" }
+author = "ysaito1001"

aws/rust-runtime/aws-inlineable/src/glacier_checksums.rs

@@ -14,7 +14,6 @@ use bytes::Buf;
 use bytes_utils::SegmentedBuf;
 use http::header::HeaderName;
 use ring::digest::{Context, Digest, SHA256};
-use tokio_stream::StreamExt;
 
 const TREE_HASH_HEADER: &str = "x-amz-sha256-tree-hash";
 const X_AMZ_CONTENT_SHA256: &str = "x-amz-content-sha256";

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

@@ -407,9 +407,10 @@ data class RuntimeType(val path: String, val dependency: RustDependency? = null)
         fun retryErrorKind(runtimeConfig: RuntimeConfig) = smithyTypes(runtimeConfig).resolve("retry::ErrorKind")
         fun eventStreamReceiver(runtimeConfig: RuntimeConfig): RuntimeType =
             smithyHttp(runtimeConfig).resolve("event_stream::Receiver")
-
         fun eventStreamSender(runtimeConfig: RuntimeConfig): RuntimeType =
             smithyHttp(runtimeConfig).resolve("event_stream::EventStreamSender")
+        fun futuresStreamCompatByteStream(runtimeConfig: RuntimeConfig): RuntimeType =
+            smithyHttp(runtimeConfig).resolve("futures_stream_adapter::FuturesStreamCompatByteStream")
 
         fun errorMetadata(runtimeConfig: RuntimeConfig) = smithyTypes(runtimeConfig).resolve("error::ErrorMetadata")
         fun errorMetadataBuilder(runtimeConfig: RuntimeConfig) =

codegen-server/python/src/main/kotlin/software/amazon/smithy/rust/codegen/server/python/smithy/protocols/PythonServerProtocolLoader.kt

@@ -61,6 +61,7 @@ class PythonServerAfterDeserializedMemberServerHttpBoundCustomization() :
         is ServerHttpBoundProtocolSection.AfterTimestampDeserializedMember -> writable {
             rust(".into()")
         }
+
         else -> emptySection
     }
 }
@@ -78,6 +79,23 @@ class PythonServerAfterDeserializedMemberHttpBindingCustomization(private val ru
     }
 }
 
+/**
+ * Customization class used to determine how serialized stream payload should be rendered for the Python server.
+ *
+ * In this customization, we do not need to wrap the payload in a new-type wrapper to enable the
+ * `futures_core::stream::Stream` trait since the payload in question has a type
+ * `aws_smithy_http_server_python::types::ByteStream` which already implements the `Stream` trait.
+ */
+class PythonServerStreamPayloadSerializerCustomization() : ServerHttpBoundProtocolCustomization() {
+    override fun section(section: ServerHttpBoundProtocolSection): Writable = when (section) {
+        is ServerHttpBoundProtocolSection.WrapStreamPayload -> writable {
+            section.params.payloadGenerator.generatePayload(this, section.params.shapeName, section.params.shape)
+        }
+
+        else -> emptySection
+    }
+}
+
 class PythonServerProtocolLoader(
     private val supportedProtocols: ProtocolMap<ServerProtocolGenerator, ServerCodegenContext>,
 ) : ProtocolLoader<ServerProtocolGenerator, ServerCodegenContext>(supportedProtocols) {
@@ -91,6 +109,7 @@ class PythonServerProtocolLoader(
                     ),
                     additionalServerHttpBoundProtocolCustomizations = listOf(
                         PythonServerAfterDeserializedMemberServerHttpBoundCustomization(),
+                        PythonServerStreamPayloadSerializerCustomization(),
                     ),
                     additionalHttpBindingCustomizations = listOf(
                         PythonServerAfterDeserializedMemberHttpBindingCustomization(runtimeConfig),
@@ -103,6 +122,7 @@ class PythonServerProtocolLoader(
                     ),
                     additionalServerHttpBoundProtocolCustomizations = listOf(
                         PythonServerAfterDeserializedMemberServerHttpBoundCustomization(),
+                        PythonServerStreamPayloadSerializerCustomization(),
                     ),
                     additionalHttpBindingCustomizations = listOf(
                         PythonServerAfterDeserializedMemberHttpBindingCustomization(runtimeConfig),
@@ -115,6 +135,7 @@ class PythonServerProtocolLoader(
                     ),
                     additionalServerHttpBoundProtocolCustomizations = listOf(
                         PythonServerAfterDeserializedMemberServerHttpBoundCustomization(),
+                        PythonServerStreamPayloadSerializerCustomization(),
                     ),
                     additionalHttpBindingCustomizations = listOf(
                         PythonServerAfterDeserializedMemberHttpBindingCustomization(runtimeConfig),

codegen-server/src/main/kotlin/software/amazon/smithy/rust/codegen/server/smithy/protocols/ServerHttpBoundProtocolGenerator.kt

@@ -81,11 +81,28 @@ import software.amazon.smithy.rust.codegen.server.smithy.generators.protocol.Ser
 import software.amazon.smithy.rust.codegen.server.smithy.generators.serverBuilderSymbol
 import java.util.logging.Logger
 
+data class StreamPayloadSerializerParams(
+    val codegenContext: ServerCodegenContext,
+    val payloadGenerator: ServerHttpBoundProtocolPayloadGenerator,
+    val shapeName: String,
+    val shape: OperationShape,
+)
+
 /**
  * Class describing a ServerHttpBoundProtocol section that can be used in a customization.
  */
 sealed class ServerHttpBoundProtocolSection(name: String) : Section(name) {
     data class AfterTimestampDeserializedMember(val shape: MemberShape) : ServerHttpBoundProtocolSection("AfterTimestampDeserializedMember")
+
+    /**
+     * Represent a section for rendering the serialized stream payload.
+     *
+     * If the payload does not implement the `futures_core::stream::Stream`, which is the case for
+     * `aws_smithy_http::byte_stream::ByteStream`, the section needs to be overridden and renders a new-type wrapper
+     * around the payload to enable the `Stream` trait.
+     */
+    data class WrapStreamPayload(val params: StreamPayloadSerializerParams) :
+        ServerHttpBoundProtocolSection("WrapStreamPayload")
 }
 
 /**
@@ -540,7 +557,18 @@ class ServerHttpBoundProtocolTraitImplGenerator(
         operationShape.outputShape(model).findStreamingMember(model)?.let {
             val payloadGenerator = ServerHttpBoundProtocolPayloadGenerator(codegenContext, protocol)
             withBlockTemplate("let body = #{SmithyHttpServer}::body::boxed(#{SmithyHttpServer}::body::Body::wrap_stream(", "));", *codegenScope) {
-                payloadGenerator.generatePayload(this, "output", operationShape)
+                for (customization in customizations) {
+                    customization.section(
+                        ServerHttpBoundProtocolSection.WrapStreamPayload(
+                            StreamPayloadSerializerParams(
+                                codegenContext,
+                                payloadGenerator,
+                                "output",
+                                operationShape,
+                            ),
+                        ),
+                    )(this)
+                }
             }
         } ?: run {
             val payloadGenerator = ServerHttpBoundProtocolPayloadGenerator(codegenContext, protocol)
@@ -707,7 +735,7 @@ class ServerHttpBoundProtocolTraitImplGenerator(
                 rustTemplate(
                     """
                     #{SmithyHttpServer}::protocol::content_type_header_classifier(
-                        &parts.headers, 
+                        &parts.headers,
                         Some("$expectedRequestContentType"),
                     )?;
                     input = #{parser}(bytes.as_ref(), input)?;

codegen-server/src/main/kotlin/software/amazon/smithy/rust/codegen/server/smithy/protocols/ServerProtocolLoader.kt

@@ -9,21 +9,68 @@ import software.amazon.smithy.aws.traits.protocols.AwsJson1_0Trait
 import software.amazon.smithy.aws.traits.protocols.AwsJson1_1Trait
 import software.amazon.smithy.aws.traits.protocols.RestJson1Trait
 import software.amazon.smithy.aws.traits.protocols.RestXmlTrait
+import software.amazon.smithy.rust.codegen.core.rustlang.Writable
+import software.amazon.smithy.rust.codegen.core.rustlang.withBlockTemplate
+import software.amazon.smithy.rust.codegen.core.rustlang.writable
+import software.amazon.smithy.rust.codegen.core.smithy.RuntimeType
 import software.amazon.smithy.rust.codegen.core.smithy.protocols.AwsJsonVersion
 import software.amazon.smithy.rust.codegen.core.smithy.protocols.ProtocolLoader
 import software.amazon.smithy.rust.codegen.core.smithy.protocols.ProtocolMap
+import software.amazon.smithy.rust.codegen.core.util.isOutputEventStream
 import software.amazon.smithy.rust.codegen.server.smithy.ServerCodegenContext
 import software.amazon.smithy.rust.codegen.server.smithy.generators.protocol.ServerProtocolGenerator
 
+class StreamPayloadSerializerCustomization() : ServerHttpBoundProtocolCustomization() {
+    override fun section(section: ServerHttpBoundProtocolSection): Writable = when (section) {
+        is ServerHttpBoundProtocolSection.WrapStreamPayload -> writable {
+            if (section.params.shape.isOutputEventStream(section.params.codegenContext.model)) {
+                // Event stream payload, of type `aws_smithy_http::event_stream::MessageStreamAdapter`, already
+                // implements the `Stream` trait, so no need to wrap it in the new-type.
+                section.params.payloadGenerator.generatePayload(this, section.params.shapeName, section.params.shape)
+            } else {
+                // Otherwise, the stream payload is `aws_smithy_http::byte_stream::ByteStream`. We wrap it in the
+                // new-type to enable the `Stream` trait.
+                withBlockTemplate(
+                    "#{FuturesStreamCompatByteStream}::new(",
+                    ")",
+                    "FuturesStreamCompatByteStream" to RuntimeType.futuresStreamCompatByteStream(section.params.codegenContext.runtimeConfig),
+                ) {
+                    section.params.payloadGenerator.generatePayload(
+                        this,
+                        section.params.shapeName,
+                        section.params.shape,
+                    )
+                }
+            }
+        }
+
+        else -> emptySection
+    }
+}
+
 class ServerProtocolLoader(supportedProtocols: ProtocolMap<ServerProtocolGenerator, ServerCodegenContext>) :
     ProtocolLoader<ServerProtocolGenerator, ServerCodegenContext>(supportedProtocols) {
 
     companion object {
         val DefaultProtocols = mapOf(
-            RestJson1Trait.ID to ServerRestJsonFactory(),
-            RestXmlTrait.ID to ServerRestXmlFactory(),
-            AwsJson1_0Trait.ID to ServerAwsJsonFactory(AwsJsonVersion.Json10),
-            AwsJson1_1Trait.ID to ServerAwsJsonFactory(AwsJsonVersion.Json11),
+            RestJson1Trait.ID to ServerRestJsonFactory(
+                additionalServerHttpBoundProtocolCustomizations = listOf(
+                    StreamPayloadSerializerCustomization(),
+                ),
+            ),
+            RestXmlTrait.ID to ServerRestXmlFactory(
+                additionalServerHttpBoundProtocolCustomizations = listOf(
+                    StreamPayloadSerializerCustomization(),
+                ),
+            ),
+            AwsJson1_0Trait.ID to ServerAwsJsonFactory(
+                AwsJsonVersion.Json10,
+                additionalServerHttpBoundProtocolCustomizations = listOf(StreamPayloadSerializerCustomization()),
+            ),
+            AwsJson1_1Trait.ID to ServerAwsJsonFactory(
+                AwsJsonVersion.Json11,
+                additionalServerHttpBoundProtocolCustomizations = listOf(StreamPayloadSerializerCustomization()),
+            ),
         )
     }
 }

codegen-server/src/main/kotlin/software/amazon/smithy/rust/codegen/server/smithy/protocols/ServerRestXmlFactory.kt

@@ -15,11 +15,17 @@ import software.amazon.smithy.rust.codegen.server.smithy.generators.protocol.Ser
  * RestXml server-side protocol factory. This factory creates the [ServerHttpProtocolGenerator]
  * with RestXml specific configurations.
  */
-class ServerRestXmlFactory : ProtocolGeneratorFactory<ServerHttpBoundProtocolGenerator, ServerCodegenContext> {
+class ServerRestXmlFactory(
+    private val additionalServerHttpBoundProtocolCustomizations: List<ServerHttpBoundProtocolCustomization> = listOf(),
+) : ProtocolGeneratorFactory<ServerHttpBoundProtocolGenerator, ServerCodegenContext> {
     override fun protocol(codegenContext: ServerCodegenContext): Protocol = ServerRestXmlProtocol(codegenContext)
 
     override fun buildProtocolGenerator(codegenContext: ServerCodegenContext): ServerHttpBoundProtocolGenerator =
-        ServerHttpBoundProtocolGenerator(codegenContext, ServerRestXmlProtocol(codegenContext))
+        ServerHttpBoundProtocolGenerator(
+            codegenContext,
+            ServerRestXmlProtocol(codegenContext),
+            additionalServerHttpBoundProtocolCustomizations,
+        )
 
     override fun support(): ProtocolSupport {
         return ProtocolSupport(

rust-runtime/aws-smithy-http-server-python/src/types.rs

@@ -31,7 +31,6 @@ use pyo3::{
     prelude::*,
 };
 use tokio::{runtime::Handle, sync::Mutex};
-use tokio_stream::StreamExt;
 
 use crate::PyError;