Provide an opt-in in aws-smithy-types-convert to enable Stream trait

[ysaito1001]

Context: #2978 (comment)

With PaginationStream no longer implementing the Stream trait, we can add a new-type for PaginationStream to the aws-smithy-types-convert crate to enable the Stream trait for that new-type.

This is an opt-in feature and improves the API ergonomics. Furthermore, when the Stream trait goes to 1.0, we can add a new supporting new-type in a backward compatible way.

[1oglop1]

How am I supposed to use aws-smithy-types-convert from my code to display RFC 3339 and ISO 8601 without directly depending on aws-smithy-types-convert?
The SDK is hiding so many things that makes work so much easier and every time I have to depend on aws-types or something else that is marked as

This crate is part of the [AWS SDK for Rust](https://awslabs.github.io/aws-sdk-rust/) and the [smithy-rs](https://github.com/awslabs/smithy-rs) code generator. In most cases, it should not be used directly.

because it is not reexported.

Then it becomes a real pain to update any of the packages in a workspace with multiple crates using workspace_hack.

[jdisanti]

The vast majority of things you need should be re-exported, and it's a bug if they're not (please file something to let us know which type you needed). However, aws-smithy-types-convert is the exception to that general rule currently. Each release of the SDK lists crate versions for that release in an expandable section to identify which versions you should use. For example, see the "Click to expand to view crate versions..." in: https://github.com/awslabs/aws-sdk-rust/releases/tag/release-2023-10-26

[1oglop1]

and it's a bug if they're not (please file something to let us know which type you need
Thanks for the response, I will move more of my questions into discussions so I do not hijack this issue.

There are a lot of crates, some of them have grouped versioning, but I saw somewhere that you are trying to line it up. ❤️

I am rooting for more examples, because standard docs.rs are not great for discovery.

Especially around the aws_sdk_config because coming from boto3 I've had a hard time creating a different config other than load_from_env for a local version of dynamoDB.
Also, fewer Option<T>s would be great.

Off topic

Coming up with this config for a client of the local version of dynamo db took me quite some time. I use this client during integration tests and just wanted to set the values.

As a newcomer to Rust I find it rather confusing, how the configuration of the SDK works (but I am comparing this to Boto3 and JS v3). Then 2 similar crates aws-sdk-config and aws-config just add to the pile of confusing information.
I am hoping, you can iron it out before v1.0 🤞 .

My app depends on and probably a lot has changed since then but I was not in a position to update due to some changes that broke the other parts of the app that required aws-types - but it looks like it is probably resolved in later versions.

aws-sdk-dynamodb = "0.28.0"
aws-sdk-ssm = "0.28.0"
aws-sdk-sqs = "0.28.0"
aws-config = "0.55.3"
aws-types = "0.55.3"

aws_lambda_events = "0.10.0"
serde_dynamo = { version = "4.2.3", features = ["aws-sdk-dynamodb+0_28"] }
aws-credential-types = { version = "0.55.3", features = ["hardcoded-credentials"] }

here's the example what I did with the config

pub fn dev_credentials() -> SdkConfig {
   use aws_credential_types::provider::SharedCredentialsProvider;
   use aws_credential_types::Credentials;
   let c = Credentials::new("x", "x", None, None, "my_provider_name");

   aws_config::SdkConfig::builder()
       .credentials_provider(SharedCredentialsProvider::new(c))
       .region(Region::new("local-env"))
       .endpoint_url("http://localhost:5001")
       .build()
}

I'd welcome more options for serialisation as mentioned in my previous comment.
serde_dynamo is IMO a must-have and I would not dare to use dynamo SDK without it because it is incredibly wordy.

// here it would be great if I could serialize it right away but for now I just printed the result or will make a dedicated struct for the output.

// 
    println!("AWS_ACCESS_KEY_ID={}",credentials.access_key_id.ok_or_else(|| eyre!("new error message"))?);
    println!("AWS_SECRET_ACCESS_KEY={}",credentials.secret_access_key.ok_or_else(|| eyre!("new error message"))?);
    println!("AWS_SESSION_TOKEN={}",credentials.session_token.ok_or_else(|| eyre!("new error message"))?);
        let expiration: aws_sdk_ecr::primitives::DateTime = credentials.expiration.ok_or_else(|| eyre!("new error message"))?;
    let expiration = expiration.to_chrono_utc()?;
    println!("AWS_CREDENTIAL_EXPIRATION={}",expiration.with_timezone(&Local).to_rfc3339_opts(chrono::SecondsFormat::Millis, true));

Lastly error handling improvements. Our codebase uses eyre::Report (anyhow) and I had problems with surfacing the actual service error up. Instead of some generic string saying "Service error".

I will try to update to the latest version as soon as possible, to see the improvements.

[jdisanti]

Thanks for the detailed feedback!

Especially around the aws_sdk_config because coming from boto3 I've had a hard time creating a different config other than load_from_env for a local version of dynamoDB.

I think what you're looking for is ConfigLoader. Example usage:

let config = aws_config::from_env()
    .endpoint_url("http://localhost:5001")
    .load()
    .await;

Then 2 similar crates aws-sdk-config and aws-config just add to the pile of confusing information.

Yeah, it's not ideal. Most people will never want aws-sdk-config unless they're using https://aws.amazon.com/config/

Not a lot we can do about this unfortunately.

Lastly error handling improvements. Our codebase uses eyre::Report (anyhow) and I had problems with surfacing the actual service error up. Instead of some generic string saying "Service error".

Take a look at DisplayErrorContext for extracting better error information. Hopefully the nightly-only methods on std::error::Error will become stable some day so that this can be improved.

[ahrens]

Would this be a reasonable way to implement the Stream trait?

struct StreamPaginationStream<I> {
    inner: PaginationStream<I>,
}

impl<I> StreamPaginationStream<I> {
    fn new(inner: PaginationStream<I>) -> Self {
        Self { inner }
    }
}

impl<I> Stream for StreamPaginationStream<I>
where
    Self: Unpin,
{
    type Item = I;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        let fut = self.inner.next();
        let fut = pin!(fut);
        fut.poll(cx)
    }
}

(plus an extension trait to create a StreamPaginationStream when doing ...into_paginator().send().into_stream())

[jdisanti]

I don't think that will quite work since if your poll_next function returns Poll::Pending, it will actually grab a new next future the next time it is called (after the waker wakes). Therefore, it would skip an item in that case, and it could even cascade if the next item isn't immediately ready.

Unfortunately, you would need to create a finite state machine to make it work.

[ahrens]

That makes sense in the general case. I think that given the implementation of PaginationStream::next(), which calls FnStream::next(), which calls FnStream::poll_next(), the future here doesn't save any state, so it will happen to work with the current implementation, but it could break if this implementation changes. I think to make it work in the general case, I need to save the Future returned by PaginationStream::next(), and keep polling the same Future until it returns Ready, at which point I can call next() again to get a different Future.

[ahrens]

I couldn't easily figure out how to save the Future, but I was able to use the genawaiter crate which I assume does this internally:

[dependencies]
genawaiter = { version = "0.99.1", features = ["futures03"] }

use genawaiter::sync::gen;
use genawaiter::yield_;
...
        let mut stream = ...into_paginator().send();
        let stream = gen!({
            while let Some(output) = stream.next().await {
                yield_!(output);
            }
        });
        // stream now implements Stream

[ahrens]

Switched to the more-used async-stream crate:

use async_stream::stream;
...
       let mut stream = ...into_paginator().send();
       let stream = stream!({
            while let Some(output) = stream.next().await {
                yield output;
            }
        });
        // stream now implements Stream

[jdisanti]

Implemented in #3299