Struct AsyncReceiver 

pub struct AsyncReceiver<T> { /* private fields */ }

AsyncReceiver is receiving side of the channel in async mode. Receivers can be cloned and produce receivers to operate in both sync and async modes.

Examples

let (_s, receiver) = kanal::bounded_async::<u64>(0);
let sync_receiver=receiver.clone_sync();

Implementations

impl<T> AsyncReceiver<T>

pub fn recv(&self) -> ReceiveFuture<'_, T>

Returns a ReceiveFuture to receive data from the channel asynchronously.

Cancellation and Polling Considerations

Due to current limitations in Rust’s handling of future cancellation, if a ReceiveFuture is dropped exactly at the time when new data is written to the channel, it may result in the loss of the received value. This behavior although memory-safe stems from the fact that Rust does not provide a built-in, correct mechanism for cancelling futures.

Additionally, it is important to note that constructs such as tokio::select! are not correct to use with kanal async channels. Kanal’s design does not rely on the conventional poll mechanism to read messages. Because of its internal optimizations, the future may complete without receiving the final poll, which prevents proper handling of the message.

As a result, once the ReceiveFuture is polled for the first time (which registers the request to receive data), the programmer must commit to completing the polling process. This ensures that messages are correctly delivered and avoids potential race conditions associated with cancellation.

Examples

let name=r.recv().await?;
println!("Hello {}",name);

Examples found in repository

examples/echo_client.rs (line 91)

    async fn run(&mut self) -> CoreResult<()> {
        loop {
            tokio::select! {
                Ok(cmd) = self.cmd_rx.recv() => {
                    let _ = self.to_ws.send(Message::text(cmd)).await;
                    self.echo.store(true, Ordering::SeqCst);
                }
                Ok(msg) = self.msg_rx.recv() => {
                    if let Message::Text(txt) = &*msg {
                        if self.echo.load(Ordering::SeqCst) {
                            let _ = self.cmd_tx.send(txt.to_string()).await;
                            self.echo.store(false, Ordering::SeqCst);
                        }
                    }
                }
            }
        }
    }

    fn rule(_: Arc<()>) -> Box<dyn Rule + Send + Sync> {
        Box::new(move |msg: &Message| msg.is_text())
    }
}

#[derive(Clone)]
pub struct EchoHandle {
    sender: AsyncSender<String>,
    receiver: AsyncReceiver<String>,
}

impl EchoHandle {
    pub async fn echo(&self, msg: String) -> CoreResult<String> {
        let _ = self.sender.send(msg).await;
        println!("In side echo handle, waiting for response...");
        Ok(self.receiver.recv().await?)
    }
}

// --- ApiModule 2: StreamModule ---
pub struct StreamModule {
    msg_rx: AsyncReceiver<Arc<Message>>,
    cmd_rx: AsyncReceiver<bool>,
    cmd_tx: AsyncSender<String>,
    send: AtomicBool,
}

#[async_trait]
impl ApiModule<()> for StreamModule {
    type Command = bool;
    type CommandResponse = String;
    type Handle = StreamHandle;

    fn new(
        _state: Arc<()>,
        cmd_rx: AsyncReceiver<Self::Command>,
        cmd_ret_tx: AsyncSender<Self::CommandResponse>,
        msg_rx: AsyncReceiver<Arc<Message>>,
        _to_ws: AsyncSender<Message>,
        _: AsyncSender<RunnerCommand>,
    ) -> Self {
        Self {
            msg_rx,
            cmd_tx: cmd_ret_tx,
            cmd_rx,
            send: AtomicBool::new(false),
        }
    }

    fn create_handle(
        sender: AsyncSender<Self::Command>,
        receiver: AsyncReceiver<Self::CommandResponse>,
    ) -> Self::Handle {
        StreamHandle { sender, receiver }
    }

    async fn run(&mut self) -> CoreResult<()> {
        loop {
            tokio::select! {
                Ok(cmd) = self.cmd_rx.recv() => {
                    // Update the send flag based on the received command
                    self.send.store(cmd, Ordering::SeqCst);
                }
                Ok(msg) = self.msg_rx.recv() => {
                    if let Message::Text(txt) = &*msg {
                        if self.send.load(Ordering::SeqCst) {
                            // Process the message if send is true
                            println!("[StreamModule] Received: {txt}");
                            let _ = self.cmd_tx.send(txt.to_string()).await;
                        }
                    }
                }
                else => {
                    println!("[Error] StreamModule: Channel closed");
                },
            }
        }
    }

    fn rule(_: Arc<()>) -> Box<dyn Rule + Send + Sync> {
        Box::new(move |_msg: &Message| {
            // Accept all messages
            true
        })
    }
}

#[derive(Clone)]
pub struct StreamHandle {
    receiver: AsyncReceiver<String>,
    sender: AsyncSender<bool>,
}

impl StreamHandle {
    pub async fn stream(self) -> CoreResult<impl Stream<Item = CoreResult<String>>> {
        self.sender.send(true).await?;
        println!("StreamHandle: Waiting for messages...");
        Ok(Box::pin(unfold(self.receiver, |state| async move {
            let item = state.recv().await.map_err(CoreError::from);
            Some((item, state))
        })))
    }
}

// --- ApiModule 3: PeriodicSenderModule ---
pub struct PeriodicSenderModule {
    cmd_rx: AsyncReceiver<bool>,
    to_ws: AsyncSender<Message>,
    running: AtomicBool,
}

#[async_trait]
impl ApiModule<()> for PeriodicSenderModule {
    type Command = bool; // true = start, false = stop
    type CommandResponse = ();
    type Handle = PeriodicSenderHandle;

    fn new(
        _state: Arc<()>,
        cmd_rx: AsyncReceiver<Self::Command>,
        _cmd_ret_tx: AsyncSender<Self::CommandResponse>,
        _msg_rx: AsyncReceiver<Arc<Message>>,
        to_ws: AsyncSender<Message>,
        _: AsyncSender<RunnerCommand>,
    ) -> Self {
        Self {
            cmd_rx,
            to_ws,
            running: AtomicBool::new(false),
        }
    }

    fn create_handle(
        sender: AsyncSender<Self::Command>,
        _receiver: AsyncReceiver<Self::CommandResponse>,
    ) -> Self::Handle {
        PeriodicSenderHandle { sender }
    }

    async fn run(&mut self) -> CoreResult<()> {
        let to_ws = self.to_ws.clone();
        let mut interval = tokio::time::interval(Duration::from_secs(5));
        loop {
            tokio::select! {
                Ok(cmd) = self.cmd_rx.recv() => {
                    self.running.store(cmd, Ordering::SeqCst);
                }
                _ = interval.tick() => {
                    if self.running.load(Ordering::SeqCst) {
                        let _ = to_ws.send(Message::text("Ping from periodic sender")).await;
                    }
                }
            }
        }
    }

examples/testing_echo_client.rs (line 81)

    async fn run(&mut self) -> CoreResult<()> {
        loop {
            tokio::select! {
                Ok(cmd) = self.cmd_rx.recv() => {
                    let _ = self.to_ws.send(Message::text(cmd)).await;
                    self.echo.store(true, Ordering::SeqCst);
                }
                Ok(msg) = self.msg_rx.recv() => {
                    if let Message::Text(txt) = &*msg {
                        if self.echo.load(Ordering::SeqCst) {
                            let _ = self.cmd_tx.send(txt.to_string()).await;
                            self.echo.store(false, Ordering::SeqCst);
                        }
                    }
                }
            }
        }
    }

    fn rule(_: Arc<()>) -> Box<dyn Rule + Send + Sync> {
        Box::new(move |msg: &Message| {
            println!("Routing rule for EchoModule: {msg:?}");
            msg.is_text()
        })
    }
}

#[derive(Clone)]
pub struct EchoHandle {
    sender: AsyncSender<String>,
    receiver: AsyncReceiver<String>,
}

impl EchoHandle {
    pub async fn echo(&self, msg: String) -> CoreResult<String> {
        let _ = self.sender.send(msg).await;
        println!("In side echo handle, waiting for response...");
        Ok(self.receiver.recv().await?)
    }

pub fn stream(&self) -> ReceiveStream<'_, T>

Creates a asynchronous stream for the channel to receive messages, ReceiveStream borrows the AsyncReceiver, after dropping it, receiver will be available and usable again.

Examples

// import to be able to use stream.next() function
use futures::stream::StreamExt;
// import to be able to use stream.is_terminated() function
use futures::stream::FusedStream;

let (s, r) = kanal::unbounded_async();
co(async move {
    for i in 0..100 {
        s.send(i).await.unwrap();
    }
});
let mut stream = r.stream();
assert!(!stream.is_terminated());
for i in 0..100 {
    assert_eq!(stream.next().await, Some(i));
}
// Stream will return None after it is terminated, and there is no other sender.
assert_eq!(stream.next().await, None);
assert!(stream.is_terminated());

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

Tries receiving from the channel without waiting on the waitlist. It returns Ok(Some(T)) in case of successful operation and Ok(None) for a failed one, or error in case that channel is closed. Important note: this function is not lock-free as it acquires a mutex guard of the channel internal for a short time.

Examples

loop {
    if let Some(name)=r.try_recv()?{
        println!("Hello {}!",name);
        break;
    }
}

pub fn try_recv_realtime(&self) -> Result<Option<T>, ReceiveError>

Tries receiving from the channel without waiting on the waitlist or waiting for channel internal lock. It returns Ok(Some(T)) in case of successful operation and Ok(None) for a failed one, or error in case that channel is closed. Do not use this function unless you know exactly what you are doing.

Examples

loop {
    if let Some(name)=r.try_recv_realtime()?{
        println!("Hello {}!",name);
        break;
    }
}

pub fn drain_into(&self, vec: &mut Vec<T>) -> Result<usize, ReceiveError>

Drains all available messages from the channel into the provided vector and returns the number of received messages.

The function is designed to be non-blocking, meaning it only processes messages that are readily available and returns immediately with whatever messages are present. It provides a count of received messages, which could be zero if no messages are available at the time of the call.

When using this function, it’s a good idea to check if the returned count is zero to avoid busy-waiting in a loop. If blocking behavior is desired when the count is zero, you can use the recv() function if count is zero. For efficiency, reusing the same vector across multiple calls can help minimize memory allocations. Between uses, you can clear the vector with vec.clear() to prepare it for the next set of messages.

Examples

let mut buf = Vec::with_capacity(1000);
loop {
    if let Ok(count) = r.drain_into(&mut buf) {
        if count == 0 {
           // count is 0, to avoid busy-wait using recv for
           // the first next message
           if let Ok(v) = r.recv() {
              buf.push(v);
           } else {
             break;
           }
        }
        // use buffer
        buf.iter().for_each(|v| println!("{}",v));
    }else{
        println!("Channel closed");
        break;
    }
    buf.clear();
}

pub fn is_disconnected(&self) -> bool

Returns, whether the send side of the channel, is closed or not.

Examples

let (s, r) = kanal::unbounded::<u64>();
drop(s); // drop sender and disconnect the send side from the channel
assert_eq!(r.is_disconnected(),true);

pub fn is_terminated(&self) -> bool

Returns, whether the channel receive side is terminated, and will not return any result in future recv calls.

Examples

let (s, r) = kanal::unbounded::<u64>();
s.send(1).unwrap();
drop(s); // drop sender and disconnect the send side from the channel
assert_eq!(r.is_disconnected(),true);
// Also channel is closed from send side, it's not terminated as there is data in channel queue
assert_eq!(r.is_terminated(),false);
assert_eq!(r.recv().unwrap(),1);
// Now channel receive side is terminated as there is no sender for channel and queue is empty
assert_eq!(r.is_terminated(),true);

pub fn clone_sync(&self) -> Receiver<T>

Returns sync cloned version of the receiver.

Examples

let (s, r) = kanal::unbounded_async();
s.send(1).await?;
let sync_receiver=r.clone_sync();
// JUST FOR EXAMPLE IT IS WRONG TO USE SYNC INSTANCE IN ASYNC CONTEXT
assert_eq!(sync_receiver.recv()?,1);

pub fn to_sync(self) -> Receiver<T>

Converts AsyncReceiver to Receiver and returns it.

Examples

  let (s, r) = kanal::bounded_async(0);
  // move to sync environment
  std::thread::spawn(move || {
    let r=r.to_sync();
    let name=r.recv()?;
    println!("Hello {}!",name);
    anyhow::Ok(())
  });
  s.send("World").await?;

pub fn as_sync(&self) -> &Receiver<T>

Borrows AsyncReceiver as Receiver and returns it

Examples

  let (s, r) = kanal::bounded_async(0);
  // move to sync environment
  std::thread::spawn(move || {
    let name=r.as_sync().recv()?;
    println!("Hello {}!",name);
    anyhow::Ok(())
  });
  s.send("World").await?;

pub fn is_bounded(&self) -> bool

Returns whether the channel is bounded or not.

Examples

let (s, r) = kanal::bounded::<u64>(0);
assert_eq!(s.is_bounded(),true);
assert_eq!(r.is_bounded(),true);

let (s, r) = kanal::unbounded::<u64>();
assert_eq!(s.is_bounded(),false);
assert_eq!(r.is_bounded(),false);

pub fn len(&self) -> usize

Returns length of the queue.

Examples

let (s, r) = kanal::unbounded::<u64>();
assert_eq!(s.len(),0);
assert_eq!(r.len(),0);
s.send(10);
assert_eq!(s.len(),1);
assert_eq!(r.len(),1);

pub fn is_empty(&self) -> bool

Returns whether the channel queue is empty or not.

Examples

let (s, r) = kanal::unbounded::<u64>();
assert_eq!(s.is_empty(),true);
assert_eq!(r.is_empty(),true);

pub fn is_full(&self) -> bool

Returns whether the channel queue is full or not full channels will block on send and recv calls it always returns true for zero sized channels.

Examples

let (s, r) = kanal::bounded(1);
s.send("Hi!").unwrap();
assert_eq!(s.is_full(),true);
assert_eq!(r.is_full(),true);

pub fn capacity(&self) -> usize

Returns capacity of channel (not the queue) for unbounded channels, it will return usize::MAX.

Examples

let (s, r) = kanal::bounded::<u64>(0);
assert_eq!(s.capacity(),0);
assert_eq!(r.capacity(),0);

let (s, r) = kanal::unbounded::<u64>();
assert_eq!(s.capacity(),usize::MAX);
assert_eq!(r.capacity(),usize::MAX);

pub fn receiver_count(&self) -> u32

Returns count of alive receiver instances of the channel.

Examples

let (s, r) = kanal::unbounded::<u64>();
let receiver_clone=r.clone();
assert_eq!(r.receiver_count(),2);

pub fn sender_count(&self) -> u32

Returns count of alive sender instances of the channel.

Examples

let (s, r) = kanal::unbounded::<u64>();
let sender_clone=s.clone();
assert_eq!(r.sender_count(),2);

pub fn close(&self) -> Result<(), CloseError>

Closes the channel completely on both sides and terminates waiting signals.

Examples

let (s, r) = kanal::unbounded::<u64>();
// closes channel on both sides and has same effect as r.close();
s.close().unwrap();
assert_eq!(r.is_closed(),true);
assert_eq!(s.is_closed(),true);

pub fn is_closed(&self) -> bool

Returns whether the channel is closed on both side of send and receive or not.

Examples

let (s, r) = kanal::unbounded::<u64>();
// closes channel on both sides and has same effect as r.close();
s.close();
assert_eq!(r.is_closed(),true);
assert_eq!(s.is_closed(),true);

Trait Implementations

impl<T> Clone for AsyncReceiver<T>

Available on crate feature async only.

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

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<T> Debug for AsyncReceiver<T>

Available on crate feature async only.

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

Formats the value using the given formatter.

impl<T> Drop for AsyncReceiver<T>

Available on crate feature async only.

fn drop(&mut self)

Executes the destructor for this type.