Remove futures_core::stream::Stream from aws_smithy_http::byte_stream::ByteStream

CHANGELOG.next.toml

@@ -150,3 +150,27 @@ message = "Fix regression with redacting sensitive HTTP response bodies."
 references = ["smithy-rs#2926", "smithy-rs#2972"]
 meta = { "breaking" = false, "tada" = false, "bug" = true, "target" = "client" }
 author = "ysaito1001"
+
+[[aws-sdk-rust]]
+message = "The `futures_core::stream::Stream` trait has been removed from public API. It should not affect usual SDK use cases. If your code uses paginators, you do not need to use the `Stream` trait or its exntension traits, but only the `next`, `try_next`, `collect` methods are supported on `PaginationStream`. Other stream operations made available through the trait or its extension traits are not supported, but we should be able to add them as needed in a backward compatible manner."
+references = ["smithy-rs#2978"]
+meta = { "breaking" = true, "tada" = false, "bug" = false }
+author = "ysaito1001"
+
+[[smithy-rs]]
+message = "The `futures_core::stream::Stream` trait has been removed from public API. [`FnStream`](https://docs.rs/aws-smithy-async/latest/aws_smithy_async/future/pagination_stream/struct.FnStream.html) only supports `next`, `try_next`, and `collect` methods. [`TryFlatMap::flat_map`](https://docs.rs/aws-smithy-async/latest/aws_smithy_async/future/pagination_stream/struct.TryFlatMap.html#method.flat_map) returns [`PaginationStream`](https://docs.rs/aws-smithy-async/latest/aws_smithy_async/future/pagination_stream/struct.PaginationStream.html), which should be preferred to `FnStream` at an interface level. For stream operations previously made available through the trait or its extension traits, we should be able to add them as needed in a backward compatible manner."
+references = ["smithy-rs#2978"]
+meta = { "breaking" = true, "tada" = false, "bug" = false, "target" = "client" }
+author = "ysaito1001"
+
+[[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 made available through the trait or its extension traits are not supported, but we should be able to add them as needed 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 made available through the trait or its extension traits are not supported, but we should be able to add them as needed 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";

aws/sdk/integration-tests/dynamodb/tests/paginators.rs

@@ -6,8 +6,6 @@
 use std::collections::HashMap;
 use std::iter::FromIterator;
 
-use tokio_stream::StreamExt;
-
 use aws_credential_types::Credentials;
 use aws_sdk_dynamodb::types::AttributeValue;
 use aws_sdk_dynamodb::{Client, Config};

aws/sdk/integration-tests/ec2/tests/paginators.rs

@@ -3,8 +3,6 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-use tokio_stream::StreamExt;
-
 use aws_sdk_ec2::{config::Credentials, config::Region, types::InstanceType, Client, Config};
 use aws_smithy_client::http_connector::HttpConnector;
 use aws_smithy_client::test_connection::TestConnection;

codegen-client/src/main/kotlin/software/amazon/smithy/rust/codegen/client/smithy/generators/PaginatorGenerator.kt

@@ -90,7 +90,7 @@ class PaginatorGenerator private constructor(
         "HttpResponse" to RuntimeType.smithyRuntimeApi(runtimeConfig).resolve("client::orchestrator::HttpResponse"),
         "SdkError" to RuntimeType.sdkError(runtimeConfig),
         "client" to RuntimeType.smithyClient(runtimeConfig),
-        "fn_stream" to RuntimeType.smithyAsync(runtimeConfig).resolve("future::fn_stream"),
+        "pagination_stream" to RuntimeType.smithyAsync(runtimeConfig).resolve("future::pagination_stream"),
 
         // External Types
         "Stream" to RuntimeType.TokioStream.resolve("Stream"),
@@ -141,13 +141,14 @@ class PaginatorGenerator private constructor(
 
                 /// Create the pagination stream
                 ///
-                /// _Note:_ No requests will be dispatched until the stream is used (eg. with [`.next().await`](tokio_stream::StreamExt::next)).
-                pub fn send(self) -> impl #{Stream}<Item = #{item_type}> + #{Unpin} {
+                /// _Note:_ No requests will be dispatched until the stream is used
+                /// (e.g. with the [`.next().await`](aws_smithy_async::future::pagination_stream::PaginationStream::next) method).
+                pub fn send(self) -> #{pagination_stream}::PaginationStream<#{item_type}> {
                     // Move individual fields out of self for the borrow checker
                     let builder = self.builder;
                     let handle = self.handle;
                     #{runtime_plugin_init}
-                    #{fn_stream}::FnStream::new(move |tx| #{Box}::pin(async move {
+                    #{pagination_stream}::PaginationStream::new(#{pagination_stream}::FnStream::new(move |tx| #{Box}::pin(async move {
                         // Build the input for the first time. If required fields are missing, this is where we'll produce an early error.
                         let mut input = match builder.build().map_err(#{SdkError}::construction_failure) {
                             #{Ok}(input) => input,
@@ -177,7 +178,7 @@ class PaginatorGenerator private constructor(
                                 return
                             }
                         }
-                    }))
+                    })))
                 }
             }
             """,
@@ -257,11 +258,12 @@ class PaginatorGenerator private constructor(
                 impl ${paginatorName}Items {
                     /// Create the pagination stream
                     ///
-                    /// _Note: No requests will be dispatched until the stream is used (eg. with [`.next().await`](tokio_stream::StreamExt::next))._
+                    /// _Note_: No requests will be dispatched until the stream is used
+                    /// (e.g. with the [`.next().await`](aws_smithy_async::future::pagination_stream::PaginationStream::next) method).
                     ///
-                    /// To read the entirety of the paginator, use [`.collect::<Result<Vec<_>, _>()`](tokio_stream::StreamExt::collect).
-                    pub fn send(self) -> impl #{Stream}<Item = #{item_type}> + #{Unpin} {
-                        #{fn_stream}::TryFlatMap::new(self.0.send()).flat_map(|page| #{extract_items}(page).unwrap_or_default().into_iter())
+                    /// To read the entirety of the paginator, use [`.collect::<Result<Vec<_>, _>()`](aws_smithy_async::future::pagination_stream::PaginationStream::collect).
+                    pub fn send(self) -> #{pagination_stream}::PaginationStream<#{item_type}> {
+                        #{pagination_stream}::TryFlatMap::new(self.0.send()).flat_map(|page| #{extract_items}(page).unwrap_or_default().into_iter())
                     }
                 }
 

codegen-client/src/main/kotlin/software/amazon/smithy/rust/codegen/client/smithy/generators/client/FluentClientGenerator.kt

@@ -433,7 +433,7 @@ class FluentClientGenerator(
                         """
                         /// Create a paginator for this request
                         ///
-                        /// Paginators are used by calling [`send().await`](#{Paginator}::send) which returns a `Stream`.
+                        /// Paginators are used by calling [`send().await`](#{Paginator}::send) which returns a [`PaginationStream`](aws_smithy_async::future::pagination_stream::PaginationStream).
                         pub fn into_paginator(self) -> #{Paginator} {
                             #{Paginator}::new(self.handle, self.inner)
                         }

codegen-client/src/main/kotlin/software/amazon/smithy/rust/codegen/client/smithy/protocols/HttpBoundProtocolGenerator.kt

@@ -7,12 +7,32 @@ package software.amazon.smithy.rust.codegen.client.smithy.protocols
 
 import software.amazon.smithy.rust.codegen.client.smithy.ClientCodegenContext
 import software.amazon.smithy.rust.codegen.core.rustlang.CargoDependency
+import software.amazon.smithy.rust.codegen.core.rustlang.RustWriter
+import software.amazon.smithy.rust.codegen.core.rustlang.rust
 import software.amazon.smithy.rust.codegen.core.rustlang.rustTemplate
 import software.amazon.smithy.rust.codegen.core.smithy.RuntimeType
 import software.amazon.smithy.rust.codegen.core.smithy.generators.http.HttpMessageType
 import software.amazon.smithy.rust.codegen.core.smithy.generators.protocol.ProtocolPayloadGenerator
 import software.amazon.smithy.rust.codegen.core.smithy.protocols.HttpBoundProtocolPayloadGenerator
 import software.amazon.smithy.rust.codegen.core.smithy.protocols.Protocol
+import software.amazon.smithy.rust.codegen.core.smithy.protocols.StreamPayloadSerializerParams
+import software.amazon.smithy.rust.codegen.core.smithy.protocols.StreamPayloadSerializerRenderer
+
+private class ClientStreamPayloadSerializerRenderer : StreamPayloadSerializerRenderer {
+    override fun renderOutputType(writer: RustWriter, params: StreamPayloadSerializerParams) {
+        writer.rust(
+            "#T",
+            RuntimeType.futuresStreamCompatByteStream(params.runtimeConfig).toSymbol(),
+        )
+    }
+
+    override fun renderPayload(writer: RustWriter, params: StreamPayloadSerializerParams) {
+        writer.rust(
+            "#T::new(${params.payloadName!!})",
+            RuntimeType.futuresStreamCompatByteStream(params.runtimeConfig),
+        )
+    }
+}
 
 class ClientHttpBoundProtocolPayloadGenerator(
     codegenContext: ClientCodegenContext,
@@ -41,4 +61,5 @@ class ClientHttpBoundProtocolPayloadGenerator(
             "errorMarshallerConstructorFn" to params.errorMarshallerConstructorFn,
         )
     },
+    ClientStreamPayloadSerializerRenderer(),
 )

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-core/src/main/kotlin/software/amazon/smithy/rust/codegen/core/smithy/protocols/HttpBoundProtocolPayloadGenerator.kt

@@ -6,6 +6,7 @@
 package software.amazon.smithy.rust.codegen.core.smithy.protocols
 
 import software.amazon.smithy.codegen.core.CodegenException
+import software.amazon.smithy.codegen.core.SymbolProvider
 import software.amazon.smithy.model.shapes.BlobShape
 import software.amazon.smithy.model.shapes.DocumentShape
 import software.amazon.smithy.model.shapes.MemberShape
@@ -17,12 +18,14 @@ import software.amazon.smithy.model.traits.EnumTrait
 import software.amazon.smithy.rust.codegen.core.rustlang.CargoDependency
 import software.amazon.smithy.rust.codegen.core.rustlang.RustWriter
 import software.amazon.smithy.rust.codegen.core.rustlang.rust
-import software.amazon.smithy.rust.codegen.core.rustlang.rustBlockTemplate
+import software.amazon.smithy.rust.codegen.core.rustlang.rustTemplate
 import software.amazon.smithy.rust.codegen.core.rustlang.withBlock
 import software.amazon.smithy.rust.codegen.core.rustlang.withBlockTemplate
 import software.amazon.smithy.rust.codegen.core.smithy.CodegenContext
 import software.amazon.smithy.rust.codegen.core.smithy.CodegenTarget
+import software.amazon.smithy.rust.codegen.core.smithy.RuntimeConfig
 import software.amazon.smithy.rust.codegen.core.smithy.RuntimeType
+import software.amazon.smithy.rust.codegen.core.smithy.RuntimeType.Companion.preludeScope
 import software.amazon.smithy.rust.codegen.core.smithy.generators.http.HttpMessageType
 import software.amazon.smithy.rust.codegen.core.smithy.generators.operationBuildError
 import software.amazon.smithy.rust.codegen.core.smithy.generators.protocol.AdditionalPayloadContext
@@ -50,11 +53,38 @@ data class EventStreamBodyParams(
     val additionalPayloadContext: AdditionalPayloadContext,
 )
 
+data class StreamPayloadSerializerParams(
+    val symbolProvider: SymbolProvider,
+    val runtimeConfig: RuntimeConfig,
+    val member: MemberShape,
+    val payloadName: String?,
+)
+
+/**
+ * An interface to help customize how to render a stream payload serializer.
+ *
+ * When the output of the serializer is passed to `hyper::body::Body::wrap_stream`,
+ * it requires what's passed to implement `futures_core::stream::Stream` trait.
+ * However, a certain type, such as `aws_smithy_http::byte_stream::ByteStream` does not
+ * implement the trait, so we need to wrap it with a new-type that does implement the trait.
+ *
+ * Each implementing type of the interface can choose whether the payload should be wrapped
+ * with such a new-type or should simply be used as-is.
+ */
+interface StreamPayloadSerializerRenderer {
+    /** Renders the return type of stream payload serializer **/
+    fun renderOutputType(writer: RustWriter, params: StreamPayloadSerializerParams)
+
+    /** Renders the stream payload **/
+    fun renderPayload(writer: RustWriter, params: StreamPayloadSerializerParams)
+}
+
 class HttpBoundProtocolPayloadGenerator(
     codegenContext: CodegenContext,
     private val protocol: Protocol,
     private val httpMessageType: HttpMessageType = HttpMessageType.REQUEST,
     private val renderEventStreamBody: (RustWriter, EventStreamBodyParams) -> Unit,
+    private val streamPayloadSerializerRenderer: StreamPayloadSerializerRenderer,
 ) : ProtocolPayloadGenerator {
     private val symbolProvider = codegenContext.symbolProvider
     private val model = codegenContext.model
@@ -63,6 +93,7 @@ class HttpBoundProtocolPayloadGenerator(
     private val httpBindingResolver = protocol.httpBindingResolver
     private val smithyEventStream = RuntimeType.smithyEventStream(runtimeConfig)
     private val codegenScope = arrayOf(
+        *preludeScope,
         "hyper" to CargoDependency.HyperWithStream.toType(),
         "SdkBody" to RuntimeType.sdkBody(runtimeConfig),
         "BuildError" to runtimeConfig.operationBuildError(),
@@ -238,17 +269,22 @@ class HttpBoundProtocolPayloadGenerator(
     ) {
         val ref = if (payloadMetadata.takesOwnership) "" else "&"
         val serializer = protocolFunctions.serializeFn(member, fnNameSuffix = "http_payload") { fnName ->
-            val outputT = if (member.isStreaming(model)) {
-                symbolProvider.toSymbol(member)
-            } else {
-                RuntimeType.ByteSlab.toSymbol()
-            }
-            rustBlockTemplate(
-                "pub fn $fnName(payload: $ref#{Member}) -> Result<#{outputT}, #{BuildError}>",
+            rustTemplate(
+                "pub(crate) fn $fnName(payload: $ref#{Member}) -> #{Result}<",
                 "Member" to symbolProvider.toSymbol(member),
-                "outputT" to outputT,
                 *codegenScope,
-            ) {
+            )
+            if (member.isStreaming(model)) {
+                streamPayloadSerializerRenderer.renderOutputType(
+                    this,
+                    StreamPayloadSerializerParams(symbolProvider, runtimeConfig, member, null),
+                )
+            } else {
+                rust("#T", RuntimeType.ByteSlab.toSymbol())
+            }
+            rustTemplate(", #{BuildError}>", *codegenScope)
+
+            withBlockTemplate("{", "}", *codegenScope) {
                 val asRef = if (payloadMetadata.takesOwnership) "" else ".as_ref()"
 
                 if (symbolProvider.toSymbol(member).isOptional()) {
@@ -303,8 +339,10 @@ class HttpBoundProtocolPayloadGenerator(
             is BlobShape -> {
                 // Write the raw blob to the payload.
                 if (member.isStreaming(model)) {
-                    // Return the `ByteStream`.
-                    rust(payloadName)
+                    streamPayloadSerializerRenderer.renderPayload(
+                        this,
+                        StreamPayloadSerializerParams(symbolProvider, runtimeConfig, member, payloadName),
+                    )
                 } else {
                     // Convert the `Blob` into a `Vec<u8>` and return it.
                     rust("$payloadName.into_inner()")

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

@@ -61,6 +61,29 @@ class PythonServerAfterDeserializedMemberServerHttpBoundCustomization() :
         is ServerHttpBoundProtocolSection.AfterTimestampDeserializedMember -> writable {
             rust(".into()")
         }
+
+        else -> emptySection
+    }
+}
+
+/**
+ * Customization class used to determine the type of serialized stream payload and how it should be wrapped in a
+ * new-type wrapper to enable `futures_core::stream::Stream` trait.
+ */
+class PythonServerStreamPayloadSerializerCustomization() : ServerHttpBoundProtocolCustomization() {
+    override fun section(section: ServerHttpBoundProtocolSection): Writable = when (section) {
+        is ServerHttpBoundProtocolSection.TypeOfSerializedStreamPayload -> writable {
+            // `aws_smithy_http_server_python::types::ByteStream` already implements
+            // `futures::stream::Stream`, so no need to wrap it in a futures' stream-compatible
+            // wrapper.
+            rust("#T", section.params.symbolProvider.toSymbol(section.params.member))
+        }
+
+        is ServerHttpBoundProtocolSection.WrapStreamPayload -> writable {
+            // payloadName is always non-null within WrapStreamAfterPayloadGenerated
+            rust(section.params.payloadName!!)
+        }
+
         else -> emptySection
     }
 }
@@ -91,6 +114,7 @@ class PythonServerProtocolLoader(
                     ),
                     additionalServerHttpBoundProtocolCustomizations = listOf(
                         PythonServerAfterDeserializedMemberServerHttpBoundCustomization(),
+                        PythonServerStreamPayloadSerializerCustomization(),
                     ),
                     additionalHttpBindingCustomizations = listOf(
                         PythonServerAfterDeserializedMemberHttpBindingCustomization(runtimeConfig),
@@ -103,6 +127,7 @@ class PythonServerProtocolLoader(
                     ),
                     additionalServerHttpBoundProtocolCustomizations = listOf(
                         PythonServerAfterDeserializedMemberServerHttpBoundCustomization(),
+                        PythonServerStreamPayloadSerializerCustomization(),
                     ),
                     additionalHttpBindingCustomizations = listOf(
                         PythonServerAfterDeserializedMemberHttpBindingCustomization(runtimeConfig),
@@ -115,6 +140,7 @@ class PythonServerProtocolLoader(
                     ),
                     additionalServerHttpBoundProtocolCustomizations = listOf(
                         PythonServerAfterDeserializedMemberServerHttpBoundCustomization(),
+                        PythonServerStreamPayloadSerializerCustomization(),
                     ),
                     additionalHttpBindingCustomizations = listOf(
                         PythonServerAfterDeserializedMemberHttpBindingCustomization(runtimeConfig),