DOCSEARCHbyAlgolia
Submit Docs

async-channel Rust Documentation

repository·master·Indexed 21 days ago

https://github.com/smol-rs/async-channel

An asynchronous multi-producer multi-consumer (MPMC) channel library for Rust (version 2.5.0). It provides both bounded and unbounded channels, allowing multiple tasks to send and receive messages. Features include asynchronous and non-blocking send/receive operations, synchronous blocking methods, Stream implementation for receivers, and support for weak references via WeakSender and WeakReceiver to manage channel lifecycles.

Tokens
2.1K
Snippets
7
Records
9
Agent score
26%

What's inside async-channel

  1. What is async-channel and how does it work?

    master

    async-channel is an asynchronous multi-producer multi-consumer (MPMC) channel implementation. In this model, each message sent into the channel is received by exactly one consumer among all active receivers.

    There are two types of channels available:

    1. Bounded channels: Have a fixed, limited capacity.
    2. Unbounded channels: Have unlimited capacity.

    A channel consists of a Sender and a Receiver side. Both sides are Clone, allowing them to be shared across multiple threads or tasks.

    Channel Lifecycle and Closing:

    • Automatic Closing: A channel closes automatically when either all Senders are dropped or all Receivers are dropped.
    • Manual Closing: You can manually close a channel by calling Sender::close() or Receiver::close().
    • Behavior when closed: Once a channel is closed, no new messages can be sent. However, any messages that were already in the channel can still be received by consumers.

  2. How channel closing and lifecycle works

    master

    A channel is considered closed when:

    1. All Senders associated with the channel are dropped.
    2. All Receivers associated with the channel are dropped.
    3. Either Sender::close() or Receiver::close() is called manually.

    Behavior when closed:

    • No more messages can be sent. Any attempt to send will return an error (SendError or TrySendError::Closed).
    • If there are messages remaining in the queue, they can still be received via recv() or try_recv().
    • Once the queue is empty and the channel is closed, recv() will return RecvError and try_recv() will return TryRecvError::Closed.

  3. Create an unbounded channel

    master

    To create a channel with unlimited capacity, use the async_channel::unbounded() function. This returns a tuple containing a Sender and a Receiver.

    let (s, r) = async_channel::unbounded();

assert_eq!(s.send("Hello").await, Ok(()));
assert_eq!(r.recv().await, Ok("Hello"));

  4. Send messages using Sender

    master

    The Sender<T> type provides several ways to push messages into the channel:

    • send(msg): An asynchronous method that waits if the channel is full. Returns a Send future.
    • try_send(msg): A non-blocking method. Returns Ok(()) if successful, Err(TrySendError::Full(msg)) if the channel is full, or Err(TrySendError::Closed(msg)) if the channel is closed.
    • send_blocking(msg): A synchronous method that blocks the current thread until the message is sent. Warning: Do not use this in an asynchronous context as it may cause deadlocks.
    • force_send(msg): A method that forcefully pushes a message. If the channel is full, it replaces an existing message and returns it as Ok(Some(old_msg)). Returns Err(SendError(msg)) if the channel is closed.
    • close(): Manually closes the channel. Remaining messages can still be received.
    // Async send
s.send(msg).await?;

// Non-blocking try_send
match s.try_send(msg) {
    Ok(_) => println!("Sent"),
    Err(TrySendError::Full(m)) => println!("Channel full, kept: {:?}", m),
    Err(TrySendError::Closed(m)) => println!("Channel closed, kept: {:?}", m),
}

// Force send (replaces existing message if full)
let result = s.force_send(msg)?;
if let Some(old_msg) = result {
    println!("Replaced: {:?}", old_msg);
}

  5. Create bounded and unbounded channels

    master

    You can create two types of channels in async-channel:

    1. Bounded channels: Created with bounded(cap), these have a fixed capacity cap. If the capacity is reached, senders will wait until space becomes available.
    2. Unbounded channels: Created with unbounded(), these have unlimited capacity and will not block senders due to fullness.

    Both types return a tuple containing a Sender<T> and a Receiver<T>. Both sides are cloneable and can be shared across multiple threads.

    // Bounded channel with capacity of 1
let (s, r) = async_channel::bounded(1);

// Unbounded channel
let (s, r) = async_channel::unbounded();

  6. Use WeakSender and WeakReceiver to avoid preventing channel closure

    master

    By default, holding a Sender or Receiver keeps the channel open. If you need to hold a reference to a channel without preventing it from closing when the primary owners are dropped, use downgrade().

    • Sender::downgrade() returns a WeakSender<T>.
    • Receiver::downgrade() returns a WeakReceiver<T>.

    To use a weak reference, you must call .upgrade(), which returns an Option<Sender<T>> or Option<Receiver<T>>. If the channel is already closed, upgrade() returns None.

    let (s, r) = async_channel::unbounded();
let weak_s = s.downgrade();

drop(s);

// Attempt to upgrade
if let Some(strong_s) = weak_s.upgrade() {
    strong_s.send(1).await;
} else {
    println!("Channel was closed");
}

  7. Wait for a channel to close with Sender::closed()

    master

    If you need to wait until a channel is closed, call Sender::closed(). This returns a Closed future that resolves to () once the channel is closed. This is useful for cleanup tasks or signaling the end of a producer's lifecycle.

    // Assuming `sender` is a `Sender<T>`
sender.closed().await;
println!("The channel is now closed.");

  8. Receive messages with Receiver::recv()

    master

    To receive a message from a channel asynchronously, call Receiver::recv(). This returns a Recv future that resolves to Ok(T) when a message is available, or Err(RecvError) if the channel has been closed.

    Note that Recv is marked with #[must_use], meaning you must .await the future for the receiving operation to actually occur.

    // Assuming `receiver` is a `Receiver<T>`
match receiver.recv().await {
    Ok(msg) => println!("Received: {:?}", msg),
    Err(RecvError) => println!("Channel closed"),
}

  9. Receive messages using Receiver

    master

    The Receiver<T> type provides several ways to pull messages from the channel:

    • recv(): An asynchronous method that waits until a message is available. Returns a Recv future.
    • try_recv(): A non-blocking method. Returns Ok(msg) if successful, Err(TryRecvError::Empty) if the channel is empty but not closed, or Err(TryRecvError::Closed) if the channel is empty and closed.
    • recv_blocking(): A synchronous method that blocks the current thread until a message is received. Warning: Do not use this in an asynchronous context.
    • Stream implementation: Receiver<T> implements futures_core::stream::Stream, allowing you to iterate over messages asynchronously.

    Closing the channel (via Sender::close() or Receiver::close()) prevents new messages from being sent, but allows existing messages to be drained.

    // Async receive
let msg = r.recv().await?;

// Non-blocking try_recv
match r.try_recv() {
    Ok(msg) => println!("Received: {:?}", msg),
    Err(TryRecvError::Empty) => println!("Empty"),
    Err(TryRecvError::Closed) => println!("Closed"),
}

// Using as a Stream
use futures_util::StreamExt;
while let Some(msg) = r.next().await {
    println!("Stream received: {:?}", msg);
}