Struct multiqueue::BroadcastReceiver
[−]
[src]
pub struct BroadcastReceiver<T: Clone> { /* fields omitted */ }
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) = multiqueue::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
Methods
impl<T: Clone> BroadcastReceiver<T>
[src]
fn try_recv(&self) -> Result<T, TryRecvError>
Tries to receive a value from the queue without blocking.
Examples:
use multiqueue::broadcast_queue; let (w, r) = broadcast_queue(10); w.try_send(1).unwrap(); assert_eq!(1, r.try_recv().unwrap());
use multiqueue::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();
fn recv(&self) -> Result<T, RecvError>
Receives a value from the queue, blocks until there is data.
Examples:
use multiqueue::broadcast_queue; let (w, r) = broadcast_queue(10); w.try_send(1).unwrap(); assert_eq!(1, r.recv().unwrap());
use multiqueue::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();
fn add_stream(&self) -> BroadcastReceiver<T>
Adds a new data stream to the queue, starting at the same position
as the BroadcastReceiver
this is being called on.
Examples
use multiqueue::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 multiqueue::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(); }
fn unsubscribe(self) -> bool
Removes the given reader from the queue subscription lib Returns true if this is the last reader in a given broadcast unit
Examples
use multiqueue::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");
fn try_iter<'a>(&'a self) -> BroadcastRefIter<'a, T>
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 multiqueue::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]
fn into_single(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 multiqueue::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> Clone for BroadcastReceiver<T>
[src]
fn clone(&self) -> BroadcastReceiver<T>
Returns a copy of the value. Read more
fn clone_from(&mut self, source: &Self)
1.0.0
Performs copy-assignment from source
. Read more
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?
fn into_iter(self) -> BroadcastIter<T>
Creates an iterator from a value. Read more
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?
fn into_iter(self) -> BroadcastRefIter<'a, T>
Creates an iterator from a value. Read more