Skip to content

Commit 45b2f80

Browse files
committed
Allow calling UnboundedReceiver::try_next after None
Do not panic. Not-panicking is equally safe, and does not have negative performance implication. It is irrelevant for `Stream` implementation to panic or not (because `Stream` behavior is unspecified after `None`), but panicking in `try_next` just complicates the interface: returned `Ok(None)` is reasonable assumption to have. Consider this use case: drain the queue on drop by performing app-specific cleanup of queued messages. The obvious implementation would be: ``` impl Drop for MyReceiverWrapper { fn drop(&mut self) { while let Ok(Some(m)) self.try_next() { cleanup(m); } } } ``` Without this change, I cannot even say for sure how this code need to be implemented to avoid panicking. E. g. is `is_closed` enough or some additional checks need to be performed?
1 parent d15ad36 commit 45b2f80

File tree

2 files changed

+15
-4
lines changed

2 files changed

+15
-4
lines changed

futures-channel/src/mpsc/mod.rs

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1020,9 +1020,6 @@ impl<T> Receiver<T> {
10201020
/// It is not recommended to call this function from inside of a future,
10211021
/// only when you've otherwise arranged to be notified when the channel is
10221022
/// no longer empty.
1023-
///
1024-
/// This function will panic if called after `try_next` or `poll_next` has
1025-
/// returned `None`.
10261023
pub fn try_next(&mut self) -> Result<Option<T>, TryRecvError> {
10271024
match self.next_message() {
10281025
Poll::Ready(msg) => {
@@ -1186,7 +1183,10 @@ impl<T> UnboundedReceiver<T> {
11861183
}
11871184

11881185
fn next_message(&mut self) -> Poll<Option<T>> {
1189-
let inner = self.inner.as_mut().expect("Receiver::next_message called after `None`");
1186+
let inner = match self.inner.as_mut() {
1187+
None => return Poll::Ready(None),
1188+
Some(inner) => inner,
1189+
};
11901190
// Pop off a message
11911191
match unsafe { inner.message_queue.pop_spin() } {
11921192
Some(msg) => {

futures-channel/tests/mpsc-close.rs

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -276,3 +276,14 @@ fn stress_try_send_as_receiver_closes() {
276276
bg.join()
277277
.expect("background thread join");
278278
}
279+
280+
#[test]
281+
fn try_next_after_none() {
282+
let (tx, mut rx) = mpsc::unbounded::<String>();
283+
// Drop the sender, close the channel.
284+
drop(tx);
285+
// Receive the end of channel.
286+
assert_eq!(Ok(None), rx.try_next().map_err(|_| ()));
287+
// None received, check we can call `try_next` again.
288+
assert_eq!(Ok(None), rx.try_next().map_err(|_| ()));
289+
}

0 commit comments

Comments
 (0)