feat(futures-util/stream): implement stream.unzip adapter

[binier]

The way this works is by putting the base stream in the Arc and sharing that Arc in UnzipLeft and UnzipRight streams.

Initially only UnzipLeft is polling the base stream, emits first tuple item as it's result and sends the other one through mpsc::channel to the UnzipRight. At the end of the stream, UnzipLeft sends Poll::Ready(None) through the channel to left the UnzipRight know stream is finished.

• In case UnzipLeft is dropped, first it drops mpsc::Sender and then wakes up the UnzipRight task, which is a way of telling to UnzipRight to take over polling the base stream (Since try_recv called by right stream will error with TryRecvError::Disconnected if sender is dropped).
• In case UnzipRight is dropped, right_waker.upgrade() will fail and return None since right_waker is a Weak type of Arc. If that happens, send simply won't happen.
• If both is dropped, base stream will drop too.

There is unsafe code when converting base stream from Pin<&mut Arc<Stream>> to Pin<&mut Stream> in order to poll it, but since we are accessing base stream only from one of the streams at a time, it's safe right?

closes: #2234

[binier]

I want to add some tests though, but not sure where to add them, since tests for example for adaptor zip is in disabled tests...

[taiki-e]

I don't think this is correct. You can call UnzipLeft(Right)::poll_next and UnzipRight(Left)::is_terminated at the same time.

[binier]

Oh right! But I can simply maintain is_done boolean in UnzipRight and set it to true if I receive Poll::Ready(None) right? For example Fuse<S> seems to do the same.

Or is there a case when: is_terminated() returns true for underline stream but it still hasn't emitted Poll::Ready(None) and won't ever emit it?

If thats the case, would UnzipLeft be dropped if it is_terminated? So would it be sufficient if is_terminated() returned:

• is_done if left is not dropped.
• stream.is_terminated() if left is dropped, since it's now safe to call that method.

[taiki-e]

Given that Iterator::unzip is like .collect::<(Vec<_>, Vec<_>)>, I would prefer to separate functionality like Shared Stream from Unzip.

I want to add some tests though, but not sure where to add them, since tests for example for adaptor zip is in disabled tests...

I recommend adding to futures/tests/stream.rs or futures/tests/stream_unzip.rs (new file).

[binier]

I recommend adding to futures/tests/stream.rs or futures/tests/stream_unzip.rs (new file).

Thanks! Will add tests there.

Given that Iterator::unzip is like .collect::<(Vec<_>, Vec<_>)>, I would prefer to separate functionality like Shared Stream from Unzip.

Hm... If I understand this right, you mean that unzip should not separate streams, but I didn't quite get the alternative. What would the api look like for the unzip that you are talking about?
EDIT: never mind... Got what you are saying. Looked again in the Iterator::unzip docs and instead of generating two iterators (like what I am doing), it simply consumes iterator and gives an output.

What do you suggest I do with the code? Should I change it into share() or fork() operator instead? If yes what should api look like?