Struct ringbuf::consumer::Consumer

pub struct Consumer<T, R: RbRef>where
    R::Rb: RbRead<T>,{ /* private fields */ }

Consumer part of ring buffer.

Mode

It can operate in immediate (by default) or postponed mode. Mode could be switched using Self::postponed/Self::into_postponed and Self::into_immediate methods.

• In immediate mode removed and inserted items are automatically synchronized with the other end.
• In postponed mode synchronization occurs only when Self::sync or Self::into_immediate is called or when Self is dropped. The reason to use postponed mode is that multiple subsequent operations are performed faster due to less frequent cache synchronization.

Implementations

impl<T, R: RbRef> Consumer<T, R>where R::Rb: RbRead<T>,

pub unsafe fn new(target: R) -> Self

Creates consumer from the ring buffer reference.

Safety

There must be only one consumer containing the same ring buffer reference.

pub fn rb(&self) -> &R::Rb

Returns reference to the underlying ring buffer.

pub fn into_rb_ref(self) -> R

Consumes self and returns underlying ring buffer reference.

pub fn postponed(&mut self) -> Consumer<T, RbWrap<RbReadCache<T, &R::Rb>>>

Returns postponed consumer that borrows Self.

pub fn into_postponed(self) -> Consumer<T, RbWrap<RbReadCache<T, R>>>

Transforms Self into postponed consumer.

pub fn capacity(&self) -> usize

Returns capacity of the ring buffer.

The capacity of the buffer is constant.

pub fn is_empty(&self) -> bool

Checks if the ring buffer is empty.

The result may become irrelevant at any time because of concurring producer activity.

Examples found in repository

examples/message.rs (line 37)

fn main() {
    let buf = HeapRb::<u8>::new(10);
    let (mut prod, mut cons) = buf.split();

    let smsg = "The quick brown fox jumps over the lazy dog";

    let pjh = thread::spawn(move || {
        println!("-> sending message: '{}'", smsg);

        let zero = [0];
        let mut bytes = smsg.as_bytes().chain(&zero[..]);
        loop {
            if prod.is_full() {
                println!("-> buffer is full, waiting");
                thread::sleep(Duration::from_millis(1));
            } else {
                let n = prod.read_from(&mut bytes, None).unwrap();
                if n == 0 {
                    break;
                }
                println!("-> {} bytes sent", n);
            }
        }

        println!("-> message sent");
    });

    let cjh = thread::spawn(move || {
        println!("<- receiving message");

        let mut bytes = Vec::<u8>::new();
        loop {
            if cons.is_empty() {
                if bytes.ends_with(&[0]) {
                    break;
                } else {
                    println!("<- buffer is empty, waiting");
                    thread::sleep(Duration::from_millis(1));
                }
            } else {
                let n = cons.write_into(&mut bytes, None).unwrap();
                println!("<- {} bytes received", n);
            }
        }

        assert_eq!(bytes.pop().unwrap(), 0);
        let msg = String::from_utf8(bytes).unwrap();
        println!("<- message received: '{}'", msg);

        msg
    });

    pjh.join().unwrap();
    let rmsg = cjh.join().unwrap();

    assert_eq!(smsg, rmsg);
}

pub fn is_full(&self) -> bool

Checks if the ring buffer is full.

pub fn len(&self) -> usize

The number of items stored in the buffer.

Actual number may be greater than the returned value because of concurring producer activity.

pub fn free_len(&self) -> usize

The number of remaining free places in the buffer.

Actual number may be less than the returned value because of concurring producer activity.

pub unsafe fn as_uninit_slices(&self) -> (&[MaybeUninit<T>], &[MaybeUninit<T>])

Provides a direct access to the ring buffer occupied memory. The difference from Self::as_slices is that this method provides slices of MaybeUninit<T>, so items may be moved out of slices.

Returns a pair of slices of stored items, the second one may be empty. Elements with lower indices in slice are older. First slice contains older items that second one.

Safety

All items are initialized. Elements must be removed starting from the beginning of first slice. When all items are removed from the first slice then items must be removed from the beginning of the second slice.

This method must be followed by Self::advance call with the number of items being removed previously as argument. No other mutating calls allowed before that.

pub unsafe fn as_mut_uninit_slices( &self ) -> (&mut [MaybeUninit<T>], &mut [MaybeUninit<T>])

Provides a direct mutable access to the ring buffer occupied memory.

Same as Self::as_uninit_slices.

Safety

See Self::as_uninit_slices.

pub unsafe fn advance(&mut self, count: usize)

Moves head target by count places.

Safety

First count items in occupied memory must be moved out or dropped.

pub fn as_slices(&self) -> (&[T], &[T])

Returns a pair of slices which contain, in order, the contents of the ring buffer.

pub fn as_mut_slices(&mut self) -> (&mut [T], &mut [T])

Returns a pair of mutable slices which contain, in order, the contents of the ring buffer.

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

Removes latest item from the ring buffer and returns it.

Returns None if the ring buffer is empty.

Examples found in repository

examples/static.rs (line 13)

fn main() {
    const RB_SIZE: usize = 1;
    let mut rb = StaticRb::<i32, RB_SIZE>::default();
    let (mut prod, mut cons) = rb.split_ref();

    assert_eq!(prod.push(123), Ok(()));
    assert_eq!(prod.push(321), Err(321));

    assert_eq!(cons.pop(), Some(123));
    assert_eq!(cons.pop(), None);
}

examples/simple.rs (line 11)

fn main() {
    let rb = HeapRb::<i32>::new(2);
    let (mut prod, mut cons) = rb.split();

    prod.push(0).unwrap();
    prod.push(1).unwrap();
    assert_eq!(prod.push(2), Err(2));

    assert_eq!(cons.pop().unwrap(), 0);

    prod.push(2).unwrap();

    assert_eq!(cons.pop().unwrap(), 1);
    assert_eq!(cons.pop().unwrap(), 2);
    assert_eq!(cons.pop(), None);
}

pub fn pop_iter(&mut self) -> PopIterator<'_, T, R>

Returns an iterator that removes items one by one from the ring buffer.

Iterator provides only items that are available for consumer at the moment of pop_iter call, it will not contain new items added after it was created.

Information about removed items is commited to the buffer only when iterator is destroyed.

pub fn iter(&self) -> impl Iterator<Item = &T> + '_

Returns a front-to-back iterator containing references to items in the ring buffer.

This iterator does not remove items out of the ring buffer.

pub fn iter_mut(&mut self) -> impl Iterator<Item = &mut T> + '_

Returns a front-to-back iterator that returns mutable references to items in the ring buffer.

This iterator does not remove items out of the ring buffer.

pub fn skip(&mut self, count: usize) -> usize

Removes at most n and at least min(n, Self::len()) items from the buffer and safely drops them.

If there is no concurring producer activity then exactly min(n, Self::len()) items are removed.

Returns the number of deleted items.

let target = HeapRb::<i32>::new(8);
let (mut prod, mut cons) = target.split();

assert_eq!(prod.push_iter(&mut (0..8)), 8);

assert_eq!(cons.skip(4), 4);
assert_eq!(cons.skip(8), 4);
assert_eq!(cons.skip(8), 0);

pub fn clear(&mut self) -> usize

Removes all items from the buffer and safely drops them.

Returns the number of deleted items.

impl<T: Copy, R: RbRef> Consumer<T, R>where R::Rb: RbRead<T>,

pub fn pop_slice(&mut self, elems: &mut [T]) -> usize

Removes first items from the ring buffer and writes them into a slice. Elements must be Copy.

Returns count of items been removed from the ring buffer.

impl<T, R: RbRef> Consumer<T, RbWrap<RbReadCache<T, R>>>where R::Rb: RbRead<T>,

pub unsafe fn new_postponed(target: R) -> Self

Create new postponed consumer.

Safety

There must be only one consumer containing the same ring buffer reference.

pub fn sync(&mut self)

Synchronize changes with the ring buffer.

Postponed consumer requires manual synchronization to make freed space visible for the producer.

pub fn into_immediate(self) -> Consumer<T, R>

Synchronize and transform back to immediate consumer.

impl<R: RbRef> Consumer<u8, R>where R::Rb: RbRead<u8>,

pub fn write_into<P: Write>( &mut self, writer: &mut P, count: Option<usize> ) -> Result<usize>

Removes at most first count bytes from the ring buffer and writes them into a Write instance. If count is None then as much as possible bytes will be written.

Returns Ok(n) if write succeeded. n is number of bytes been written. n == 0 means that either write returned zero or ring buffer is empty.

If write is failed then original error is returned. In this case it is guaranteed that no items was written to the writer. To achieve this we write only one contiguous slice at once. So this call may write less than len items even if the writer is ready to get more.

Examples found in repository

examples/message.rs (line 45)

fn main() {
    let buf = HeapRb::<u8>::new(10);
    let (mut prod, mut cons) = buf.split();

    let smsg = "The quick brown fox jumps over the lazy dog";

    let pjh = thread::spawn(move || {
        println!("-> sending message: '{}'", smsg);

        let zero = [0];
        let mut bytes = smsg.as_bytes().chain(&zero[..]);
        loop {
            if prod.is_full() {
                println!("-> buffer is full, waiting");
                thread::sleep(Duration::from_millis(1));
            } else {
                let n = prod.read_from(&mut bytes, None).unwrap();
                if n == 0 {
                    break;
                }
                println!("-> {} bytes sent", n);
            }
        }

        println!("-> message sent");
    });

    let cjh = thread::spawn(move || {
        println!("<- receiving message");

        let mut bytes = Vec::<u8>::new();
        loop {
            if cons.is_empty() {
                if bytes.ends_with(&[0]) {
                    break;
                } else {
                    println!("<- buffer is empty, waiting");
                    thread::sleep(Duration::from_millis(1));
                }
            } else {
                let n = cons.write_into(&mut bytes, None).unwrap();
                println!("<- {} bytes received", n);
            }
        }

        assert_eq!(bytes.pop().unwrap(), 0);
        let msg = String::from_utf8(bytes).unwrap();
        println!("<- message received: '{}'", msg);

        msg
    });

    pjh.join().unwrap();
    let rmsg = cjh.join().unwrap();

    assert_eq!(smsg, rmsg);
}

Trait Implementations

impl<R: RbRef> Read for Consumer<u8, R>where R::Rb: RbRead<u8>,

fn read(&mut self, buffer: &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, Global>) -> 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 Selfwhere 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.