[−][src]Struct multiqueue2::BroadcastReceiver
This class is the receiving half of the broadcast MultiQueue
.
Within each stream, it supports both single and multi consumer modes
with competitive performance in each case. It supports blocking and
nonblocking read modes as well as being the conduit for adding
new streams.
Examples
use std::thread; let (send, recv) = multiqueue2::broadcast_queue(4); let mut handles = vec![]; for i in 0..2 { // or n let cur_recv = recv.add_stream(); for j in 0..2 { let stream_consumer = cur_recv.clone(); handles.push(thread::spawn(move || { for val in stream_consumer { println!("Stream {} consumer {} got {}", i, j, val); } })); } // cur_recv is dropped here } // Take notice that I drop the reader - this removes it from // the queue, meaning that the readers in the new threads // won't get starved by the lack of progress from recv recv.unsubscribe(); for i in 0..10 { // Don't do this busy loop in real stuff unless you're really sure loop { if send.try_send(i).is_ok() { break; } } } drop(send); for t in handles { t.join(); } // prints along the lines of // Stream 0 consumer 1 got 2 // Stream 0 consumer 0 got 0 // Stream 1 consumer 0 got 0 // Stream 0 consumer 1 got 1 // Stream 1 consumer 1 got 1 // Stream 1 consumer 0 got 2 // etc
Implementations
impl<T: Clone> BroadcastReceiver<T>
[src]
pub fn try_recv(&self) -> Result<T, TryRecvError>
[src]
Tries to receive a value from the queue without blocking.
Examples:
use multiqueue2::broadcast_queue; let (w, r) = broadcast_queue(10); w.try_send(1).unwrap(); assert_eq!(1, r.try_recv().unwrap());
use multiqueue2::broadcast_queue; use std::thread; let (send, recv) = broadcast_queue(10); let handle = thread::spawn(move || { for _ in 0..10 { loop { match recv.try_recv() { Ok(val) => { println!("Got {}", val); break; }, Err(_) => (), } } } assert!(recv.try_recv().is_err()); // recv would block here }); for i in 0..10 { send.try_send(i).unwrap(); } // Drop the sender to close the queue drop(send); handle.join();
pub fn recv(&self) -> Result<T, RecvError>
[src]
Receives a value from the queue, blocks until there is data.
Examples:
use multiqueue2::broadcast_queue; let (w, r) = broadcast_queue(10); w.try_send(1).unwrap(); assert_eq!(1, r.recv().unwrap());
use multiqueue2::broadcast_queue; use std::thread; let (send, recv) = broadcast_queue(10); let handle = thread::spawn(move || { // note the lack of dealing with failed reads. // unwrap 'ignores' the error where sender disconnects for _ in 0..10 { println!("Got {}", recv.recv().unwrap()); } assert!(recv.try_recv().is_err()); }); for i in 0..10 { send.try_send(i).unwrap(); } // Drop the sender to close the queue drop(send); handle.join();
pub fn add_stream(&self) -> BroadcastReceiver<T>
[src]
Adds a new data stream to the queue, starting at the same position
as the BroadcastReceiver
this is being called on.
Examples
use multiqueue2::broadcast_queue; let (w, r) = broadcast_queue(10); w.try_send(1).unwrap(); assert_eq!(r.recv().unwrap(), 1); w.try_send(1).unwrap(); let r2 = r.add_stream(); assert_eq!(r.recv().unwrap(), 1); assert_eq!(r2.recv().unwrap(), 1); assert!(r.try_recv().is_err()); assert!(r2.try_recv().is_err());
use multiqueue2::broadcast_queue; use std::thread; let (send, recv) = broadcast_queue(4); let mut handles = vec![]; for i in 0..2 { // or n let cur_recv = recv.add_stream(); handles.push(thread::spawn(move || { for val in cur_recv { println!("Stream {} got {}", i, val); } })); } // Take notice that I drop the reader - this removes it from // the queue, meaning that the readers in the new threads // won't get starved by the lack of progress from recv recv.unsubscribe(); for i in 0..10 { // Don't do this busy loop in real stuff unless you're really sure loop { if send.try_send(i).is_ok() { break; } } } // Drop the sender to close the queue drop(send); for t in handles { t.join(); }
pub fn unsubscribe(self) -> bool
[src]
Removes the given reader from the queue subscription lib Returns true if this is the last reader in a given broadcast unit
Examples
use multiqueue2::broadcast_queue; let (writer, reader) = broadcast_queue(1); let reader_2_1 = reader.add_stream(); let reader_2_2 = reader_2_1.clone(); writer.try_send(1).expect("This will succeed since queue is empty"); reader.try_recv().expect("This reader can read"); assert!(writer.try_send(1).is_err(), "This fails since the reader2 group hasn't advanced"); assert!(!reader_2_2.unsubscribe(), "This returns false since reader_2_1 is still alive"); assert!(reader_2_1.unsubscribe(), "This returns true since there are no readers alive in the reader_2_x group"); writer.try_send(1).expect("This succeeds since the reader_2 group is not blocking");
pub fn try_iter(&self) -> BroadcastRefIter<'_, T>
[src]
Returns a non-owning iterator that iterates over the queue until it fails to receive an item, either through being empty or begin disconnected. This iterator will never block.
Examples:
use multiqueue2::broadcast_queue; let (w, r) = broadcast_queue(2); for _ in 0 .. 3 { w.try_send(1).unwrap(); w.try_send(2).unwrap(); for val in r.try_iter().zip(1..2) { assert_eq!(val.0, val.1); } }
impl<T: Clone + Sync> BroadcastReceiver<T>
[src]
pub fn into_single(
self
) -> Result<BroadcastUniReceiver<T>, BroadcastReceiver<T>>
[src]
self
) -> Result<BroadcastUniReceiver<T>, BroadcastReceiver<T>>
If there is only one BroadcastReceiver
on the stream, converts the
Receiver into a BroadcastUniReceiver
otherwise returns the Receiver.
Example:
use multiqueue2::broadcast_queue; let (w, r) = broadcast_queue(10); w.try_send(1).unwrap(); let r2 = r.clone(); // Fails since there's two receivers on the stream assert!(r2.into_single().is_err()); let single_r = r.into_single().unwrap(); let val = match single_r.try_recv_view(|x| 2 * *x) { Ok(val) => val, Err(_) => panic!("Queue should have an element"), }; assert_eq!(2, val);
Trait Implementations
impl<T: Clone> Clone for BroadcastReceiver<T>
[src]
pub fn clone(&self) -> BroadcastReceiver<T>
[src]
pub fn clone_from(&mut self, source: &Self)
1.0.0[src]
impl<T: Debug + Clone> Debug for BroadcastReceiver<T>
[src]
impl<T: Clone> IntoIterator for BroadcastReceiver<T>
[src]
type Item = T
The type of the elements being iterated over.
type IntoIter = BroadcastIter<T>
Which kind of iterator are we turning this into?
pub fn into_iter(self) -> BroadcastIter<T>
[src]
impl<'a, T: Clone + 'a> IntoIterator for &'a BroadcastReceiver<T>
[src]
type Item = T
The type of the elements being iterated over.
type IntoIter = BroadcastRefIter<'a, T>
Which kind of iterator are we turning this into?
pub fn into_iter(self) -> BroadcastRefIter<'a, T>
[src]
impl<T: Send + Sync + Clone> Send for BroadcastReceiver<T>
[src]
Auto Trait Implementations
impl<T> !RefUnwindSafe for BroadcastReceiver<T>
[src]
impl<T> !Sync for BroadcastReceiver<T>
[src]
impl<T> Unpin for BroadcastReceiver<T>
[src]
impl<T> !UnwindSafe for BroadcastReceiver<T>
[src]
Blanket Implementations
impl<T> Any for T where
T: 'static + ?Sized,
[src]
T: 'static + ?Sized,
impl<T> Borrow<T> for T where
T: ?Sized,
[src]
T: ?Sized,
impl<T> BorrowMut<T> for T where
T: ?Sized,
[src]
T: ?Sized,
pub fn borrow_mut(&mut self) -> &mut T
[src]
impl<T> From<T> for T
[src]
impl<T, U> Into<U> for T where
U: From<T>,
[src]
U: From<T>,
impl<T> Pointable for T
pub const ALIGN: usize
type Init = T
The type for initializers.
pub unsafe fn init(init: <T as Pointable>::Init) -> usize
pub unsafe fn deref<'a>(ptr: usize) -> &'a T
pub unsafe fn deref_mut<'a>(ptr: usize) -> &'a mut T
pub unsafe fn drop(ptr: usize)
impl<T> ToOwned for T where
T: Clone,
[src]
T: Clone,
type Owned = T
The resulting type after obtaining ownership.
pub fn to_owned(&self) -> T
[src]
pub fn clone_into(&self, target: &mut T)
[src]
impl<T, U> TryFrom<U> for T where
U: Into<T>,
[src]
U: Into<T>,
type Error = Infallible
The type returned in the event of a conversion error.
pub fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>
[src]
impl<T, U> TryInto<U> for T where
U: TryFrom<T>,
[src]
U: TryFrom<T>,