flume

To use Flume in your Rust project, add it to your Cargo.toml under the [dependencies] section.

flume = "x.y"

Flume has several optional features that can be enabled in your Cargo.toml. To use specific features, set default-features = false and list the desired features in the features array.

Available features:

• spin: Uses spinlocks instead of OS-level synchronization primitives for some internal data access. May improve performance on specific platforms/workloads.
• select: Enables the Selector API, allowing a thread to wait on multiple channels or operations simultaneously.
• async: Enables the async API, allowing you to use asynchronous operations even on synchronous channels.
• eventual-fairness: Uses randomness in the Selector implementation to prevent biasing or saturating specific events.

flume = { version = "x.y", default-features = false, features = ["async", "select"] }

The Selector type allows a thread to wait on multiple blocking operations (like receiving from multiple receivers or sending to multiple senders) simultaneously, similar to the Unix select system call.

To use it, create a new Selector, chain recv or send calls to register operations, and then call a wait method to block until one of the operations completes.

Note: If the eventual-fairness feature is enabled, the selector will pick a random ready event to prevent starvation.

let (tx0, rx0) = flume::unbounded();
let (tx1, rx1) = flume::unbounded();

std::thread::spawn(move || {
    tx0.send(true).unwrap();
    tx1.send(42).unwrap();
});

flume::Selector::new()
    .recv(&rx0, |b| println!("Received {:?}", b))
    .recv(&rx1, |n| println!("Received {:?}", n))
    .wait();

Flume provides a multi-producer, multi-consumer (MPMC) channel. You can create an unbounded channel using flume::unbounded(), which returns a (Sender, Receiver) pair. The Sender and Receiver both implement Send + Sync + Clone.

use std::thread;

fn main() {
    println!("Hello, world!");

    let (tx, rx) = flume::unbounded();

    thread::spawn(move || {
        (0..10).for_each(|i| {
            tx.send(i).unwrap();
        })
    });

    let received: u32 = rx.iter().sum();

    assert_eq!((0..10).sum::<u32>(), received);
}

Use flume::unbounded() to create a multi-producer, multi-consumer channel with infinite capacity. Returns a (Sender<T>, Receiver<T>) tuple.

let (tx, rx) = flume::unbounded();

tx.send(42).unwrap();
assert_eq!(rx.recv().unwrap(), 42);

Once you have configured a Selector, use one of the following methods to block the current thread:

• wait(): Blocks until one of the registered events completes. Panics if no events were registered.
• wait_timeout(dur: Duration): Blocks until an event completes or the specified duration expires. Returns Err(SelectError::Timeout) if no event occurs.
• wait_deadline(deadline: Instant): Blocks until an event completes or the specified deadline is reached. Returns Err(SelectError::Timeout) if no event occurs.

The Sender<T> handle allows sending values into the channel. It supports blocking sends, non-blocking attempts, and timed sends.

• send(msg): Blocks if the channel is bounded and full until space is available or all receivers are dropped.
• try_send(msg): Returns immediately. Returns TrySendError::Full if the channel is full or TrySendError::Disconnected if all receivers are dropped.
• send_timeout(msg, duration): Blocks until space is available, the timeout expires, or all receivers are dropped.
• send_deadline(msg, deadline): Blocks until space is available, the deadline is reached, or all receivers are dropped.

Adds a send operation to the selector. You provide a reference to a Sender<U>, the message msg to send, and a mapper function. The mapper function is called with a Result<(), SendError<U>> when the operation completes, and it must return a type T.

Signature: pub fn send<U, F: FnMut(Result<(), SendError<U>>) -> T + 'a>(mut self, sender: &'a Sender<U>, msg: U, mapper: F) -> Self

Adds a receive operation to the selector. You provide a reference to a Receiver<U> and a mapper function. The mapper function is called with a Result<U, RecvError> when the operation completes, and it must return a type T (the type the Selector will eventually yield).

Signature: pub fn recv<U, F: FnMut(Result<U, RecvError>) -> T + 'a>(mut self, receiver: &'a Receiver<U>, mapper: F) -> Self

Create a bounded channel with a specific maximum capacity using bounded(cap).

• If the channel reaches its capacity, calls to Sender::send will block until a receiver makes space.
• To avoid blocking, use Sender::try_send.
• Rendezvous Channels: If you provide a capacity of 0, the channel acts as a rendezvous channel, where senders block until a receiver is available to perform a handshake and transfer ownership.

Both Sender and Receiver are thread-safe and can be cloned.

let (tx, rx) = flume::bounded(32);

for i in 1..33 {
    tx.send(i).unwrap();
}
assert!(tx.try_send(33).is_err());

assert_eq!(rx.try_iter().sum::<u32>(), (1..33).sum());