Struct UnboundedReceiver 

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

Receives values from the associated UnboundedSenders.

Created by unbounded_channel. There is exactly one receiver.

Progress

Wait-free: is_closed, is_empty, len, sender_strong_count and sender_weak_count: each is a load.

Lock-free: recv, recv_many, try_recv and the two poll_ forms. Registering the waker is wait-free (the cell runs two read-modify-writes and no loop). Taking a value off the message queue is a compare-and-swap no other thread contends for, and never waits for a sender: a value whose send claimed its place and has not written it yet is not there, and its sender wakes the receiver once it is (see the Progress notes of the module). close is lock-free: one fetch_or on the count, then the set of the event that closed waits on, which takes that event’s drain role. The drop is the same as a receive per value left, since it drops every value still queued, and so takes time proportional to them.

blocking_recv and blocking_recv_many park the calling thread until a value arrives.

Implementations

impl<T> UnboundedReceiver<T>

pub async fn recv(&mut self) -> Option<T>

Receives the next value.

Returns None once the channel is closed and no buffered value is left, which means no value can ever be received again. The channel is closed when every sender has been dropped, or when close is called. If the channel is empty and open, it waits for a send or a close.

Cancel safety

Cancel safe. If the future is dropped (for instance when another branch of a select wins), no value was taken from the channel.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::unbounded_channel();
tx.send("hello").unwrap();
tx.send("world").unwrap();
drop(tx);
assert_eq!(now(rx.recv()), Some("hello"));
assert_eq!(now(rx.recv()), Some("world"));
assert_eq!(now(rx.recv()), None);

pub async fn recv_many(&mut self, buffer: &mut Vec<T>, limit: usize) -> usize

Receives the next values and appends them to buffer, taking at most limit of them in one call.

Returns how many values it added. A limit of zero returns 0 at once. Otherwise it waits like recv until at least one value is there, then takes everything that is ready up to limit without waiting again, and returns 0 only when the channel is closed and empty. The capacity of buffer grows as needed.

Cancel safety

Cancel safe, as recv: a dropped future took nothing.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::unbounded_channel();
for word in ["first", "second", "third"] {
    tx.send(word).unwrap();
}
let mut buffer = Vec::new();
assert_eq!(now(rx.recv_many(&mut buffer, 2)), 2);
assert_eq!(buffer, ["first", "second"]);
assert_eq!(now(rx.recv_many(&mut buffer, 2)), 1);
drop(tx);
assert_eq!(now(rx.recv_many(&mut buffer, 2)), 0);

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

Tries to receive the next value without waiting.

Errors

TryRecvError::Empty if the channel is empty but a sender still exists, so a value may yet come. TryRecvError::Disconnected if it is empty and nothing can send: every sender is gone, or the channel was closed.

A value whose send is still in progress on another thread is not in the channel yet, nor are the values sent after it, which come out behind it. This call does not wait for them: the sender wakes the receiver once its value is in.

use nuclei::sync::mpsc::{self, error::TryRecvError};

let (tx, mut rx) = mpsc::unbounded_channel();
tx.send("hello").unwrap();
assert_eq!(rx.try_recv(), Ok("hello"));
assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
drop(tx);
assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected));

pub fn blocking_recv(&mut self) -> Option<T>

Receives the next value from a thread that may block.

Like recv, but it parks the calling thread while the channel is empty and open. On a nuclei shard (inside one of its tasks) it does not park the thread: it runs a nested block_on, so the shard keeps running its other tasks meanwhile. Inside another runtime’s task it blocks that task’s thread.

Panics

On wasm32, where no other thread can send, when the channel is empty and open: the call could never end.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::unbounded_channel::<u8>();
let receiver = std::thread::spawn(move || {
    assert_eq!(rx.blocking_recv(), Some(10));
    assert_eq!(rx.blocking_recv(), None);
});
tx.send(10).unwrap();
drop(tx);
receiver.join().unwrap();

pub fn blocking_recv_many(&mut self, buffer: &mut Vec<T>, limit: usize) -> usize

recv_many for a thread that may block; see blocking_recv.

Panics

As for blocking_recv.

pub fn close(&mut self)

Closes the receiving half without dropping it.

No further value can be sent: a send fails and returns its value. The values already buffered stay receivable, so a clean shutdown closes, then calls recv until it returns None. A send that was already under way when the channel closed still completes and is received first.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::unbounded_channel();
tx.send(1).unwrap();
rx.close();
assert_eq!(tx.send(2).unwrap_err().0, 2);
assert_eq!(rx.try_recv(), Ok(1), "the buffered value is still there");

pub fn is_closed(&self) -> bool

Whether the channel is closed: close was called, or every sender is gone. Buffered values may still be received.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::unbounded_channel::<()>();
assert!(!rx.is_closed());
rx.close();
assert!(rx.is_closed());
drop(tx);

pub fn is_empty(&self) -> bool

Whether the channel holds no value.

use nuclei::sync::mpsc;

let (tx, rx) = mpsc::unbounded_channel();
assert!(rx.is_empty());
tx.send(()).unwrap();
assert!(!rx.is_empty());

pub fn len(&self) -> usize

How many values the channel holds.

A snapshot: it is exact while no other thread is sending, and while one is it may be off by the values being sent at that moment.

use nuclei::sync::mpsc;

let (tx, rx) = mpsc::unbounded_channel();
assert_eq!(rx.len(), 0);
tx.send(()).unwrap();
assert_eq!(rx.len(), 1);

pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>>

Polls to receive the next value.

Returns Poll::Pending if no value is available and the channel is open; Poll::Ready(Some(value)) if one is; and Poll::Ready(None) if the channel is closed and every value sent before it closed was received.

On Pending, the waker of cx is woken when a value arrives or the channel closes. Only the waker of the most recent call to poll_recv or poll_recv_many is kept.

pub fn poll_recv_many( &mut self, cx: &mut Context<'_>, buffer: &mut Vec<T>, limit: usize, ) -> Poll<usize>

Polls to receive several values into buffer, at most limit.

Returns Poll::Pending if no value is available and the channel is open; Poll::Ready(count) with the number of values added, which is at least one unless limit is zero or the channel is closed and empty. It takes everything that is ready, up to limit, in one call. The waker of cx is handled as for poll_recv.

pub fn sender_strong_count(&self) -> usize

How many UnboundedSender handles exist.

pub fn sender_weak_count(&self) -> usize

How many WeakUnboundedSender handles exist.

Trait Implementations

impl<T> Debug for UnboundedReceiver<T>

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

Formats the value using the given formatter.

impl<T> Drop for UnboundedReceiver<T>

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.