Futures-rs: TryStream bound for SinkExt::send_all

[vskh]

I noticed that recently send_all signature was updated to require TryStream bound on passed stream (https://github.com/rust-lang/futures-rs/pull/1946).

I wonder if there is a reason for this restriction?

Consider the following code that I have:

type TestData = HashMap<String, usize>;

let sink: Sink<TestData> = ...;
let msg1 = HashMap::new();
let msg2 = HashMap::new();

let mut msgs = futures::stream::iter(vec![msg1, msg2]);

sender.send_all(&mut msgs).await?;

This fails with following error:

error[E0271]: type mismatch resolving `<futures_util::stream::iter::Iter<std::vec::IntoIter<std::collections::HashMap<std::string::String, usize>>> as futures_core::stream::Stream>::Item == std::result::Result<_, _>`
  --> examples/basic.rs:55:12
   |
55 |     sender.send_all(&mut msgs).await?;
   |            ^^^^^^^^ expected struct `std::collections::HashMap`, found enum `std::result::Result`
   |
   = note: expected type `std::collections::HashMap<std::string::String, usize>`
              found enum `std::result::Result<_, _>`
   = note: required because of the requirements on the impl of `futures_core::stream::TryStream` for `futures_util::stream::iter::Iter<std::vec::IntoIter<std::collections::HashMap<std::string::String, usize>>>`

Took me some time to decrypt that this happens because stream::iter::Iter does not implement TryStream and gets automatic implementation only if stream's item is std::result::Result. So, in simple code above I have to wrap each message into Ok like futures::stream::iter(vec![Ok(msg1), Ok(msg2)]) which looks weird.

Shouldn't it be possible to send streams that produce items for sure and can't fail? Unless I am missing something obvious that makes it undesirable/not possible, I think sinking collection of values is rather often use case.

[benesch]

Yeah, this is related to an issue I filed about a month ago (#2004) and haven't heard back about. Basically I think the commit that introduced this behavior, #1946, isn't quite right. What is currently called send_all should be moved to TryStreamExt as try_send_all, and a new function send_all should be added that operates on a normal stream, not a try stream.

This is actually exactly what was proposed in #1817, but for reasons that are not clear to me it's not what was implemented in #1946.

As I mentioned before I'd be happy to try to fix this, but this project hasn't been too active of late, and I don't want to go down the wrong road. @cramertj, as the original author of #1946, can you weigh in? I'm worried I might be missing some context. Thanks!

[cramertj]

What is currently called send_all should be moved to TryStreamExt as try_send_all, and a new function send_all should be added that operates on a normal stream, not a try stream.

None of these are methods on a Stream or a TryStream-- they're methods on Sink. Sink could offer both send_all and try_send_all, but IMO there's little reason to have a non-erroring version since a Sink can always error, and it's easy to write let mut st = st.map(Ok); sink.send_all(&mut st) in the cases where the stream doesn't return Result.

[benesch]

Ah yeah, sorry, I conflated forward—which _is_ implemented on StreamExt—with send_all. I do still think it would be useful to have a version of send_all that accepts a Stream instead of a TryStream—perhaps send_all_ok would be a better name than try_send_all, since it's not the _infallible_ version of send_all, but a version of send_all that accepts an infallible stream.