Allow !Unpin Sinks in Forward combinator

[ebkalderon]

Changed

• Update Forward so it drops the sink without returning it. Users wishing to access the sink again can pass a &mut sink to StreamExt::forward().
• Update docs for both Forward and StreamExt::forward().

Removed

• Remove requirement for Unpin sinks on Forward.

Fixes #1440. /CC @Nemo157.

[Nemo157]

LGTM.

We discussed changing Forward to drop the sink on completion instead of returning it on Discord, I feel it better fits the design of the 0.3 combinators (passing a sink by-ref in an async fn if necessary), and helps with cases where you accidentally keep it alive longer than you mean to (like the fanout test does, causing the timeout on the first commit).

[ebkalderon]

@cramertj Any particular thoughts on this change to the API?

[cramertj]

It seems like maybe we should require passing Pin<&mut impl Sink> rather than taking a sink by-value, since the default behavior of forwarding and then dropping without closing will always be wrong (right?).

[Nemo157]

Another alternative would be to just update the documentation to say this will close the provided Sink once the Stream is complete (and fix the mpsc::Sender to not kill the whole channel when closed like it currently does). You could then use something like an unclosable adaptor on a Sink that just flushes but doesn't actually close it if you want to keep it active after using forward.

[ebkalderon]

@Nemo157 That sounds like a good idea. I can update the documentation, if you would like. Could you please elaborate on what you have in mind with an uncloseable adapter? Would this be another combinator we would add to the SinkExt trait?

[ebkalderon]

Another thing to note: if we do not need to return the sink directly from Forward, we can also remove the Unpin sink requirement mentioned here.

[Nemo157]

I've started the mpsc changes in #1443.

The uncloseable adaptor would just wrap a sink and call poll_flush from its poll_close, so once the adaptor was closed the underlying sink would be flushed but left open, you could then use it like

let mut sink = foo();
await!(stream.forward(sink.by_ref().dont_close()))?;
await!(sink.send(some_final_item))?;
await!(sink.close())?;

[ebkalderon]

Sounds good! I've just pushed some commits updating the Forward docs and removing the Unpin sink bound as well. I would appreciate a soundness review of my changes here.

[ebkalderon]

Pinging @cramertj @Nemo157. Do the docs and the removal of the Unpin sink bound from the last two commits look correct to you?

[ebkalderon]

@cramertj @Nemo157 Now that #1443 has been merged, I've updated my PR to switch back to using poll_close() over poll_flush(), and I have confirmed that the example code between 0.3.0 and 0.1.25 now behaves correctly. Ready for review again!

[ebkalderon]

Interesting, futures-core-preview no longer compiles on some toolchains after my rebase. Will need to look into it further.

[Nemo157]

That’s #1453, there are new changes in nightly that are waiting on #1445

[ebkalderon]

Ah, yes. I just noticed this as well when trying to navigate to std::task::LocalWaker on the Rust API docs page and it wasn't found. Standing by until #1445 gets merged.

[ebkalderon]

@cramertj @Nemo157 Awesome, looks like CI passes again thanks to #1445 being merged. 🎉 Anyone available for a final review of this PR?

[cramertj]

LGTM-- would you mind flattening your commits to remove the previous changes to Forward that have since been removed?

[ebkalderon]

@cramertj Okay, I've flattened away the redundant commits! I hope the commit history looks satisfactory now.

[cramertj]

Thanks!