Stream implementation (wrapper) for PaginationStream

[Ploppz]

Motivation and Context

awslabs/aws-sdk-rust#995

Description

I tried to implement futures::Stream for a wrapper struct around PaginationStream. I am unsure if I did it in the best way. After fighting with the borrow checker for a while I decided to try Arc<Mutex<_>> - is this the way to go or does there exist a better way?
Even then, does the code look correct? I used it in my project and my integration tests do pass but I am not 100% that these tests will catch any error in paginated ListObjectsV2.

I would appreciate any feedback so far.

Testing

In progress while waiting for feedback on code

Checklist

• I have updated CHANGELOG.next.toml if I made changes to the smithy-rs codegen or 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.

[rcoh]

for a number of reasons, I think that is the best you can do. Since you need to box the future, you then can't know for sure that the stream is alive long enough.

Luckily, there is a better way! We can expose poll_next on PaginationStream so that we can delegate to that. I've updated your branch with these changes.

I think there are a couple of remaining items we should do:

1. Add an extension trait so you can do something like:

s3.list_buckets().into_paginator().send().as_stream_03x().map(...) // and other stream trait useful functions

2. Add a changelog entry so folks know this new functionality is available!

[Ploppz]

Thanks! Some questions:

• Why do you use tokio-stream rather than futures? I guess I will change the name of the module futures.rs
• What is 03x in the function name?
• I assume at this point the code is so simply tests are not needed

[rcoh]

Thanks! Some questions:

• Why do you use tokio-stream rather than futures? I guess I will change the name of the module futures.rs

• no specific reason—it's a re-export but has Tokio's StreamExt which has a lot of useful combinators.

• What is 03x in the function name?

• The stream trait is in futures core=0.3 — although this is a good point, it should probably match the version of the tokio-stream library, 0.1, so 01x. We use this naming convention on pre-1.0 libraries so that if another version of the library is released we can create a matching method.

• I assume at this point the code is so simply tests are not needed
  Yeah. Although it couldn't hurt to validate that you read the same items with and without it, or even just validate that things compile as one would expect.

[Ploppz]

Does it look good now?

[Ploppz]

Btw, if tokio-stream re-exports futures::Stream and we don't even use StreamExt for anything, then that means we simply bring in more dependencies than necessary. Shouldn't we just use futures directly?

[abusch]

Nice, that would be useful to have! In the meantime we've done something pretty similar, which seems to work for us (plus an extension trait to add a .into_stream() method):

broken code

pub struct PaginationStreamAdapter<T>(PaginationStream<T>);

impl<T> futures::stream::Stream for PaginationStreamAdapter<T> {
    type Item = T;

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

[Ploppz]

@abusch I'm not sure but isn't it wrong to call next() in every call to poll_next?
Even if it somehow works, it seems wrong to me. I might be wrong though but here's my line of thought:
The async system will keep polling to check the status of whether the next item is ready, and every time it does that, internally, your impl will call the next() function which I believe (these semantics would make sense to me) should be waited for completion until it's called the next time. If you are lucky, it's always ready when polled, maybe that's why it works for you?

[rcoh]

FYI I'm pretty sure that implementation is wrong — It may work due to a coincidence of how our demand driven paginator works but I wouldn't recommend relying on that

…

On Thu, Dec 14, 2023, 2:35 AM Antoine Büsch ***@***.***> wrote: Nice, that would be useful to have! In the meantime we've done something pretty similar, which seems to work for us (plus an extension trait to add a .into_stream() method): pub struct PaginationStreamAdapter<T>(PaginationStream<T>); impl<T> futures::stream::Stream for PaginationStreamAdapter<T> { type Item = T; fn poll_next( mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>, ) -> std::task::Poll<Option<Self::Item>> { let fut = self.0.next(); pin_mut!(fut); fut.poll(cx) }} — Reply to this email directly, view it on GitHub <#3299 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AADYKZ37BBIWTBU75TEYIU3YJKT3ZAVCNFSM6AAAAABAMQI67CVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMYTQNJVGMZDINRSHA> . You are receiving this because your review was requested.Message ID: ***@***.***>

[abusch]

Hmm, yeah, you might be right 😓

[rcoh]

yeah that's fair. In that case, I think we should use future_core stream directly: https://docs.rs/futures-core/latest/futures_core/stream/trait.Stream.html (futures::Stream) is just yet another re-export of this.

We should probably include a comment suggesting that tokio_stream is where most of the useful utilities live, though

[rcoh]

this is looking great! just a couple of suggestions then we'll get this merged. Thanks for working on this, I think customers are really going to appreciate it.

[Ploppz]

I implemented the suggested changes, and rebased onto latest upstream main branch to fix conflicts. Also removed a trait bound that was unnecessary. Couldn't write very helpful code in the rustdoc unfortunately because we don't have a lot of dependencies to work with there.