Struct BroadcastReceiver

pub struct BroadcastReceiver<T: Clone> { /* private fields */ }

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>

pub fn try_recv(&self) -> Result<T, TryRecvError>

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>

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>

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

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>

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>

pub 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 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> Clone for BroadcastReceiver<T>

fn clone(&self) -> BroadcastReceiver<T>

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<T: Debug + Clone> Debug for BroadcastReceiver<T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<'a, T: Clone + 'a> IntoIterator for &'a BroadcastReceiver<T>

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.

impl<T: Clone> IntoIterator for BroadcastReceiver<T>

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.

impl<T: Send + Sync + Clone> Send for BroadcastReceiver<T>