mirrors/ntex
1
0
Fork 0
mirror of https://github.com/ntex-rs/ntex.git synced 2025-04-05 05:47:40 +03:00
Code Issues Projects Releases Packages Wiki Activity

Add FusedStream trait on Receiver<T> (#235)

Browse source
This commit is contained in:
tt rt 2023-11-02 10:22:45 +01:00 • committed by GitHub
parent a222316da6
commit 5c17aa1f99
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 11 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

12
ntex-util/src/channel/mpsc.rs
View file

@ -1,7 +1,7 @@
//! A multi-producer, single-consumer, futures-aware, FIFO queue. //! A multi-producer, single-consumer, futures-aware, FIFO queue.
use std::{collections::VecDeque, fmt, pin::Pin, task::Context, task::Poll}; use std::{collections::VecDeque, fmt, pin::Pin, task::Context, task::Poll};
use futures_core::Stream; use futures_core::{FusedStream, Stream};
use futures_sink::Sink; use futures_sink::Sink;
use super::cell::{Cell, WeakCell}; use super::cell::{Cell, WeakCell};
@ -206,6 +206,12 @@ impl<T> Stream for Receiver<T> {
} }
} }
impl<T> FusedStream for Receiver<T> {
fn is_terminated(&self) -> bool {
self.is_closed()
}
}
impl<T> Drop for Receiver<T> { impl<T> Drop for Receiver<T> {
fn drop(&mut self) { fn drop(&mut self) {
let shared = self.shared.get_mut(); let shared = self.shared.get_mut();
@ -313,10 +319,12 @@ mod tests {
let (tx, rx) = channel::<()>(); let (tx, rx) = channel::<()>();
assert!(!tx.is_closed()); assert!(!tx.is_closed());
assert!(!rx.is_closed()); assert!(!rx.is_closed());
assert!(!rx.is_terminated());
tx.close(); tx.close();
assert!(tx.is_closed()); assert!(tx.is_closed());
assert!(rx.is_closed()); assert!(rx.is_closed());
assert!(rx.is_terminated());
let (tx, rx) = channel::<()>(); let (tx, rx) = channel::<()>();
rx.close(); rx.close();
@ -325,7 +333,9 @@ mod tests {
let (tx, rx) = channel::<()>(); let (tx, rx) = channel::<()>();
drop(tx); drop(tx);
assert!(rx.is_closed()); assert!(rx.is_closed());
assert!(rx.is_terminated());
let _tx = rx.sender(); let _tx = rx.sender();
assert!(!rx.is_closed()); assert!(!rx.is_closed());
assert!(!rx.is_terminated());
} }
} }