Struct rtrb::Consumer

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

The consumer side of a RingBuffer.

Can be moved between threads, but references from different threads are not allowed (i.e. it is Send but not Sync).

Can only be created with RingBuffer::new() (together with its counterpart, the Producer).

Individual elements can be moved out of the ring buffer with Consumer::pop(), multiple elements at once can be read with Consumer::read_chunk().

The number of slots currently available for reading can be obtained with Consumer::slots().

When the Consumer is dropped, Producer::is_abandoned() will return true. This can be used as a crude way to communicate to the sending thread that no more data will be consumed. When the Consumer is dropped after the Producer has already been dropped, RingBuffer::drop() will be called, freeing the allocated memory.

Implementations

impl<T> Consumer<T>

pub fn read_chunk(&mut self, n: usize) -> Result<ReadChunk<'_, T>, ChunkError>

Returns n slots for reading.

ReadChunk::as_slices() provides immutable access to the slots. After reading from those slots, they explicitly have to be made available to be written again by the Producer by calling ReadChunk::commit() or ReadChunk::commit_all().

Alternatively, items can be moved out of the ReadChunk using iteration because it implements IntoIterator (ReadChunk::into_iter() can be used to explicitly turn it into an Iterator). All moved items are automatically made available to be written again by the Producer.

Errors

If not enough slots are available, an error (containing the number of available slots) is returned. Use Consumer::slots() to obtain the number of available slots beforehand.

Examples

See the documentation of the chunks module.

impl<T> Consumer<T>

pub fn pop(&mut self) -> Result<T, PopError>

Attempts to pop an element from the queue.

The element is moved out of the ring buffer and its slot is made available to be filled by the Producer again.

Errors

If the queue is empty, an error is returned.

Examples

use rtrb::{PopError, RingBuffer};

let (mut p, mut c) = RingBuffer::new(1);

assert_eq!(p.push(10), Ok(()));
assert_eq!(c.pop(), Ok(10));
assert_eq!(c.pop(), Err(PopError::Empty));

To obtain an Option<T>, use .ok() on the result.

assert_eq!(p.push(20), Ok(()));
assert_eq!(c.pop().ok(), Some(20));

pub fn peek(&self) -> Result<&T, PeekError>

Attempts to read an element from the queue without removing it.

Errors

If the queue is empty, an error is returned.

Examples

use rtrb::{PeekError, RingBuffer};

let (mut p, c) = RingBuffer::new(1);

assert_eq!(c.peek(), Err(PeekError::Empty));
assert_eq!(p.push(10), Ok(()));
assert_eq!(c.peek(), Ok(&10));
assert_eq!(c.peek(), Ok(&10));

pub fn slots(&self) -> usize

Returns the number of slots available for reading.

Since items can be concurrently produced on another thread, the actual number of available slots may increase at any time (up to the RingBuffer::capacity()).

To check for a single available slot, using Consumer::is_empty() is often quicker (because it might not have to check an atomic variable).

Examples

use rtrb::RingBuffer;

let (p, c) = RingBuffer::<f32>::new(1024);

assert_eq!(c.slots(), 0);

pub fn is_empty(&self) -> bool

Returns true if there are currently no slots available for reading.

An empty ring buffer might cease to be empty at any time if the corresponding Producer is producing items in another thread.

Examples

use rtrb::RingBuffer;

let (p, c) = RingBuffer::<f32>::new(1);

assert!(c.is_empty());

Since items can be concurrently produced on another thread, the ring buffer might not be empty for long:

if c.is_empty() {
    // The buffer might be empty, but it might as well not be
    // if an item was just produced on another thread.
}

However, if it’s not empty, another thread cannot change that:

if !c.is_empty() {
    // At least one slot is guaranteed to be available for reading.
}

pub fn is_abandoned(&self) -> bool

Returns true if the corresponding Producer has been destroyed.

Note that since Rust version 1.74.0, this is not synchronizing with the producer thread anymore, see https://github.com/mgeier/rtrb/issues/114. In a future version of rtrb, the synchronizing behavior might be restored.

Examples

use rtrb::RingBuffer;

let (mut p, mut c) = RingBuffer::new(7);
assert!(!c.is_abandoned());
assert_eq!(p.push(10), Ok(()));
drop(p);
assert!(c.is_abandoned());
// The items that are left in the ring buffer can still be consumed:
assert_eq!(c.pop(), Ok(10));

Since the producer can be concurrently dropped on another thread, the consumer might become abandoned at any time:

if !c.is_abandoned() {
    // Right now, the producer might still be alive, but it might as well not be
    // if another thread has just dropped it.
}

However, if it already is abandoned, it will stay that way:

if c.is_abandoned() {
    // This is needed since Rust 1.74.0, see https://github.com/mgeier/rtrb/issues/114:
    std::sync::atomic::fence(std::sync::atomic::Ordering::Acquire);
    // The producer does definitely not exist anymore.
}

pub fn buffer(&self) -> &RingBuffer<T>

Returns a read-only reference to the ring buffer.

Trait Implementations

impl<T: Debug> Debug for Consumer<T>

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

Formats the value using the given formatter.

impl<T: PartialEq> PartialEq for Consumer<T>

fn eq(&self, other: &Consumer<T>) -> bool

This method tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

This method tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl Read for Consumer<u8>

fn read(&mut self, buf: &mut [u8]) -> Result<usize>

Pull some bytes from this source into the specified buffer, returning how many bytes were read.

fn read_vectored(&mut self, bufs: &mut [IoSliceMut<'_>]) -> Result<usize, Error>

Like read, except that it reads into a slice of buffers.

fn is_read_vectored(&self) -> bool

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

Determines if this Reader has an efficient read_vectored implementation.

fn read_to_end(&mut self, buf: &mut Vec<u8>) -> Result<usize, Error>

Read all bytes until EOF in this source, placing them into buf.

fn read_to_string(&mut self, buf: &mut String) -> Result<usize, Error>

Read all bytes until EOF in this source, appending them to buf.

fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), Error>

Read the exact number of bytes required to fill buf.

fn read_buf(&mut self, buf: BorrowedCursor<'_>) -> Result<(), Error>

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

Pull some bytes from this source into the specified buffer.

fn read_buf_exact(&mut self, cursor: BorrowedCursor<'_>) -> Result<(), Error>

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

Read the exact number of bytes required to fill cursor.

fn by_ref(&mut self) -> &mut Self

where Self: Sized,

Creates a “by reference” adaptor for this instance of Read.

fn bytes(self) -> Bytes<Self>

where Self: Sized,

Transforms this Read instance to an Iterator over its bytes.

fn chain<R>(self, next: R) -> Chain<Self, R>

where R: Read, Self: Sized,

Creates an adapter which will chain this stream with another.

fn take(self, limit: u64) -> Take<Self>

where Self: Sized,

Creates an adapter which will read at most limit bytes from it.

impl<T: Eq> Eq for Consumer<T>

impl<T: Send> Send for Consumer<T>

impl<T> StructuralPartialEq for Consumer<T>