Struct tokio::net::unix::pipe::Receiver

source ·
pub struct Receiver { /* private fields */ }
Expand description

Reading end of a Unix pipe.

It can be constructed from a FIFO file with OpenOptions::open_receiver.

§Examples

Receiving messages from a named pipe in a loop:

use tokio::net::unix::pipe;
use tokio::io::{self, AsyncReadExt};

const FIFO_NAME: &str = "path/to/a/fifo";

let mut rx = pipe::OpenOptions::new().open_receiver(FIFO_NAME)?;
loop {
    let mut msg = vec![0; 256];
    match rx.read_exact(&mut msg).await {
        Ok(_) => {
            /* handle the message */
        }
        Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
            // Writing end has been closed, we should reopen the pipe.
            rx = pipe::OpenOptions::new().open_receiver(FIFO_NAME)?;
        }
        Err(e) => return Err(e.into()),
    }
}

On Linux, you can use a Receiver in read-write access mode to implement resilient reading from a named pipe. Unlike Receiver opened in read-only mode, read from a pipe in read-write mode will not fail with UnexpectedEof when the writing end is closed. This way, a Receiver can asynchronously wait for the next writer to open the pipe.

You should not use functions waiting for EOF such as read_to_end with a Receiver in read-write access mode, since it may wait forever. Receiver in this mode also holds an open writing end, which prevents receiving EOF.

To set the read-write access mode you can use OpenOptions::read_write. Note that using read-write access mode with FIFO files is not defined by the POSIX standard and it is only guaranteed to work on Linux.

use tokio::net::unix::pipe;
use tokio::io::AsyncReadExt;

const FIFO_NAME: &str = "path/to/a/fifo";

let mut rx = pipe::OpenOptions::new()
    .read_write(true)
    .open_receiver(FIFO_NAME)?;
loop {
    let mut msg = vec![0; 256];
    rx.read_exact(&mut msg).await?;
    /* handle the message */
}

Implementations§

source§

impl Receiver

source

pub fn from_file(file: File) -> Result<Receiver>

Creates a new Receiver from a File.

This function is intended to construct a pipe from a File representing a special FIFO file. It will check if the file is a pipe and has read access, set it in non-blocking mode and perform the conversion.

§Errors

Fails with io::ErrorKind::InvalidInput if the file is not a pipe or it does not have read access. Also fails with any standard OS error if it occurs.

§Panics

This function panics if it is not called from within a runtime with IO enabled.

The runtime is usually set implicitly when this function is called from a future driven by a tokio runtime, otherwise runtime can be set explicitly with Runtime::enter function.

source

pub fn from_owned_fd(owned_fd: OwnedFd) -> Result<Receiver>

Creates a new Receiver from an OwnedFd.

This function is intended to construct a pipe from an OwnedFd representing an anonymous pipe or a special FIFO file. It will check if the file descriptor is a pipe and has read access, set it in non-blocking mode and perform the conversion.

§Errors

Fails with io::ErrorKind::InvalidInput if the file descriptor is not a pipe or it does not have read access. Also fails with any standard OS error if it occurs.

§Panics

This function panics if it is not called from within a runtime with IO enabled.

The runtime is usually set implicitly when this function is called from a future driven by a tokio runtime, otherwise runtime can be set explicitly with Runtime::enter function.

source

pub fn from_file_unchecked(file: File) -> Result<Receiver>

Creates a new Receiver from a File without checking pipe properties.

This function is intended to construct a pipe from a File representing a special FIFO file. The conversion assumes nothing about the underlying file; it is left up to the user to make sure it is opened with read access, represents a pipe and is set in non-blocking mode.

§Examples
use tokio::net::unix::pipe;
use std::fs::OpenOptions;
use std::os::unix::fs::{FileTypeExt, OpenOptionsExt};

const FIFO_NAME: &str = "path/to/a/fifo";

let file = OpenOptions::new()
    .read(true)
    .custom_flags(libc::O_NONBLOCK)
    .open(FIFO_NAME)?;
if file.metadata()?.file_type().is_fifo() {
    let rx = pipe::Receiver::from_file_unchecked(file)?;
    /* use the Receiver */
}
§Panics

This function panics if it is not called from within a runtime with IO enabled.

The runtime is usually set implicitly when this function is called from a future driven by a tokio runtime, otherwise runtime can be set explicitly with Runtime::enter function.

source

pub fn from_owned_fd_unchecked(owned_fd: OwnedFd) -> Result<Receiver>

Creates a new Receiver from an OwnedFd without checking pipe properties.

This function is intended to construct a pipe from an OwnedFd representing an anonymous pipe or a special FIFO file. The conversion assumes nothing about the underlying pipe; it is left up to the user to make sure that the file descriptor represents the reading end of a pipe and the pipe is set in non-blocking mode.

§Panics

This function panics if it is not called from within a runtime with IO enabled.

The runtime is usually set implicitly when this function is called from a future driven by a tokio runtime, otherwise runtime can be set explicitly with Runtime::enter function.

source

pub async fn ready(&self, interest: Interest) -> Result<Ready>

Waits for any of the requested ready states.

This function can be used instead of readable() to check the returned ready set for Ready::READABLE and Ready::READ_CLOSED events.

The function may complete without the pipe being ready. This is a false-positive and attempting an operation will return with io::ErrorKind::WouldBlock. The function can also return with an empty Ready set, so you should always check the returned value and possibly wait again if the requested states are not set.

§Cancel safety

This method is cancel safe. Once a readiness event occurs, the method will continue to return immediately until the readiness event is consumed by an attempt to read that fails with WouldBlock or Poll::Pending.

source

pub async fn readable(&self) -> Result<()>

Waits for the pipe to become readable.

This function is equivalent to ready(Interest::READABLE) and is usually paired with try_read().

§Examples
use tokio::net::unix::pipe;
use std::io;

#[tokio::main]
async fn main() -> io::Result<()> {
    // Open a reading end of a fifo
    let rx = pipe::OpenOptions::new().open_receiver("path/to/a/fifo")?;

    let mut msg = vec![0; 1024];

    loop {
        // Wait for the pipe to be readable
        rx.readable().await?;

        // Try to read data, this may still fail with `WouldBlock`
        // if the readiness event is a false positive.
        match rx.try_read(&mut msg) {
            Ok(n) => {
                msg.truncate(n);
                break;
            }
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                continue;
            }
            Err(e) => {
                return Err(e.into());
            }
        }
    }

    println!("GOT = {:?}", msg);
    Ok(())
}
source

pub fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<Result<()>>

Polls for read readiness.

If the pipe is not currently ready for reading, this method will store a clone of the Waker from the provided Context. When the pipe becomes ready for reading, Waker::wake will be called on the waker.

Note that on multiple calls to poll_read_ready or poll_read, only the Waker from the Context passed to the most recent call is scheduled to receive a wakeup.

This function is intended for cases where creating and pinning a future via readable is not feasible. Where possible, using readable is preferred, as this supports polling from multiple tasks at once.

§Return value

The function returns:

  • Poll::Pending if the pipe is not ready for reading.
  • Poll::Ready(Ok(())) if the pipe is ready for reading.
  • Poll::Ready(Err(e)) if an error is encountered.
§Errors

This function may encounter any standard I/O error except WouldBlock.

source

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

Tries to read data from the pipe into the provided buffer, returning how many bytes were read.

Reads any pending data from the pipe but does not wait for new data to arrive. On success, returns the number of bytes read. Because try_read() is non-blocking, the buffer does not have to be stored by the async task and can exist entirely on the stack.

Usually readable() is used with this function.

§Return

If data is successfully read, Ok(n) is returned, where n is the number of bytes read. If n is 0, then it can indicate one of two scenarios:

  1. The pipe’s writing end is closed and will no longer write data.
  2. The specified buffer was 0 bytes in length.

If the pipe is not ready to read data, Err(io::ErrorKind::WouldBlock) is returned.

§Examples
use tokio::net::unix::pipe;
use std::io;

#[tokio::main]
async fn main() -> io::Result<()> {
    // Open a reading end of a fifo
    let rx = pipe::OpenOptions::new().open_receiver("path/to/a/fifo")?;

    let mut msg = vec![0; 1024];

    loop {
        // Wait for the pipe to be readable
        rx.readable().await?;

        // Try to read data, this may still fail with `WouldBlock`
        // if the readiness event is a false positive.
        match rx.try_read(&mut msg) {
            Ok(n) => {
                msg.truncate(n);
                break;
            }
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                continue;
            }
            Err(e) => {
                return Err(e.into());
            }
        }
    }

    println!("GOT = {:?}", msg);
    Ok(())
}
source

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

Tries to read data from the pipe into the provided buffers, returning how many bytes were read.

Data is copied to fill each buffer in order, with the final buffer written to possibly being only partially filled. This method behaves equivalently to a single call to try_read() with concatenated buffers.

Reads any pending data from the pipe but does not wait for new data to arrive. On success, returns the number of bytes read. Because try_read_vectored() is non-blocking, the buffer does not have to be stored by the async task and can exist entirely on the stack.

Usually, readable() is used with this function.

§Return

If data is successfully read, Ok(n) is returned, where n is the number of bytes read. Ok(0) indicates the pipe’s writing end is closed and will no longer write data. If the pipe is not ready to read data Err(io::ErrorKind::WouldBlock) is returned.

§Examples
use tokio::net::unix::pipe;
use std::io;

#[tokio::main]
async fn main() -> io::Result<()> {
    // Open a reading end of a fifo
    let rx = pipe::OpenOptions::new().open_receiver("path/to/a/fifo")?;

    loop {
        // Wait for the pipe to be readable
        rx.readable().await?;

        // Creating the buffer **after** the `await` prevents it from
        // being stored in the async task.
        let mut buf_a = [0; 512];
        let mut buf_b = [0; 1024];
        let mut bufs = [
            io::IoSliceMut::new(&mut buf_a),
            io::IoSliceMut::new(&mut buf_b),
        ];

        // Try to read data, this may still fail with `WouldBlock`
        // if the readiness event is a false positive.
        match rx.try_read_vectored(&mut bufs) {
            Ok(0) => break,
            Ok(n) => {
                println!("read {} bytes", n);
            }
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                continue;
            }
            Err(e) => {
                return Err(e.into());
            }
        }
    }

    Ok(())
}
source

pub fn try_read_buf<B: BufMut>(&self, buf: &mut B) -> Result<usize>

Tries to read data from the pipe into the provided buffer, advancing the buffer’s internal cursor, returning how many bytes were read.

Reads any pending data from the pipe but does not wait for new data to arrive. On success, returns the number of bytes read. Because try_read_buf() is non-blocking, the buffer does not have to be stored by the async task and can exist entirely on the stack.

Usually, readable() or ready() is used with this function.

§Return

If data is successfully read, Ok(n) is returned, where n is the number of bytes read. Ok(0) indicates the pipe’s writing end is closed and will no longer write data. If the pipe is not ready to read data Err(io::ErrorKind::WouldBlock) is returned.

§Examples
use tokio::net::unix::pipe;
use std::io;

#[tokio::main]
async fn main() -> io::Result<()> {
    // Open a reading end of a fifo
    let rx = pipe::OpenOptions::new().open_receiver("path/to/a/fifo")?;

    loop {
        // Wait for the pipe to be readable
        rx.readable().await?;

        let mut buf = Vec::with_capacity(4096);

        // Try to read data, this may still fail with `WouldBlock`
        // if the readiness event is a false positive.
        match rx.try_read_buf(&mut buf) {
            Ok(0) => break,
            Ok(n) => {
                println!("read {} bytes", n);
            }
            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                continue;
            }
            Err(e) => {
                return Err(e.into());
            }
        }
    }

    Ok(())
}
source

pub fn into_blocking_fd(self) -> Result<OwnedFd>

Converts the pipe into an OwnedFd in blocking mode.

This function will deregister this pipe end from the event loop, set it in blocking mode and perform the conversion.

source

pub fn into_nonblocking_fd(self) -> Result<OwnedFd>

Converts the pipe into an OwnedFd in nonblocking mode.

This function will deregister this pipe end from the event loop and perform the conversion. Returned file descriptor will be in nonblocking mode.

Trait Implementations§

source§

impl AsFd for Receiver

source§

fn as_fd(&self) -> BorrowedFd<'_>

Borrows the file descriptor. Read more
source§

impl AsRawFd for Receiver

source§

fn as_raw_fd(&self) -> RawFd

Extracts the raw file descriptor. Read more
source§

impl AsyncRead for Receiver

source§

fn poll_read( self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>, ) -> Poll<Result<()>>

Attempts to read from the AsyncRead into buf. Read more
source§

impl Debug for Receiver

source§

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

Formats the value using the given formatter. Read more

Auto Trait Implementations§

Blanket Implementations§

source§

impl<T> Any for T
where T: 'static + ?Sized,

source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
source§

impl<R> AsyncReadExt for R
where R: AsyncRead + ?Sized,

source§

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

Creates a new AsyncRead instance that chains this stream with next. Read more
source§

fn read<'a>(&'a mut self, buf: &'a mut [u8]) -> Read<'a, Self>
where Self: Unpin,

Pulls some bytes from this source into the specified buffer, returning how many bytes were read. Read more
source§

fn read_buf<'a, B>(&'a mut self, buf: &'a mut B) -> ReadBuf<'a, Self, B>
where Self: Unpin, B: BufMut + ?Sized,

Pulls some bytes from this source into the specified buffer, advancing the buffer’s internal cursor. Read more
source§

fn read_exact<'a>(&'a mut self, buf: &'a mut [u8]) -> ReadExact<'a, Self>
where Self: Unpin,

Reads the exact number of bytes required to fill buf. Read more
source§

fn read_u8(&mut self) -> ReadU8<&mut Self>
where Self: Unpin,

Reads an unsigned 8 bit integer from the underlying reader. Read more
source§

fn read_i8(&mut self) -> ReadI8<&mut Self>
where Self: Unpin,

Reads a signed 8 bit integer from the underlying reader. Read more
source§

fn read_u16(&mut self) -> ReadU16<&mut Self>
where Self: Unpin,

Reads an unsigned 16-bit integer in big-endian order from the underlying reader. Read more
source§

fn read_i16(&mut self) -> ReadI16<&mut Self>
where Self: Unpin,

Reads a signed 16-bit integer in big-endian order from the underlying reader. Read more
source§

fn read_u32(&mut self) -> ReadU32<&mut Self>
where Self: Unpin,

Reads an unsigned 32-bit integer in big-endian order from the underlying reader. Read more
source§

fn read_i32(&mut self) -> ReadI32<&mut Self>
where Self: Unpin,

Reads a signed 32-bit integer in big-endian order from the underlying reader. Read more
source§

fn read_u64(&mut self) -> ReadU64<&mut Self>
where Self: Unpin,

Reads an unsigned 64-bit integer in big-endian order from the underlying reader. Read more
source§

fn read_i64(&mut self) -> ReadI64<&mut Self>
where Self: Unpin,

Reads an signed 64-bit integer in big-endian order from the underlying reader. Read more
source§

fn read_u128(&mut self) -> ReadU128<&mut Self>
where Self: Unpin,

Reads an unsigned 128-bit integer in big-endian order from the underlying reader. Read more
source§

fn read_i128(&mut self) -> ReadI128<&mut Self>
where Self: Unpin,

Reads an signed 128-bit integer in big-endian order from the underlying reader. Read more
source§

fn read_f32(&mut self) -> ReadF32<&mut Self>
where Self: Unpin,

Reads an 32-bit floating point type in big-endian order from the underlying reader. Read more
source§

fn read_f64(&mut self) -> ReadF64<&mut Self>
where Self: Unpin,

Reads an 64-bit floating point type in big-endian order from the underlying reader. Read more
source§

fn read_u16_le(&mut self) -> ReadU16Le<&mut Self>
where Self: Unpin,

Reads an unsigned 16-bit integer in little-endian order from the underlying reader. Read more
source§

fn read_i16_le(&mut self) -> ReadI16Le<&mut Self>
where Self: Unpin,

Reads a signed 16-bit integer in little-endian order from the underlying reader. Read more
source§

fn read_u32_le(&mut self) -> ReadU32Le<&mut Self>
where Self: Unpin,

Reads an unsigned 32-bit integer in little-endian order from the underlying reader. Read more
source§

fn read_i32_le(&mut self) -> ReadI32Le<&mut Self>
where Self: Unpin,

Reads a signed 32-bit integer in little-endian order from the underlying reader. Read more
source§

fn read_u64_le(&mut self) -> ReadU64Le<&mut Self>
where Self: Unpin,

Reads an unsigned 64-bit integer in little-endian order from the underlying reader. Read more
source§

fn read_i64_le(&mut self) -> ReadI64Le<&mut Self>
where Self: Unpin,

Reads an signed 64-bit integer in little-endian order from the underlying reader. Read more
source§

fn read_u128_le(&mut self) -> ReadU128Le<&mut Self>
where Self: Unpin,

Reads an unsigned 128-bit integer in little-endian order from the underlying reader. Read more
source§

fn read_i128_le(&mut self) -> ReadI128Le<&mut Self>
where Self: Unpin,

Reads an signed 128-bit integer in little-endian order from the underlying reader. Read more
source§

fn read_f32_le(&mut self) -> ReadF32Le<&mut Self>
where Self: Unpin,

Reads an 32-bit floating point type in little-endian order from the underlying reader. Read more
source§

fn read_f64_le(&mut self) -> ReadF64Le<&mut Self>
where Self: Unpin,

Reads an 64-bit floating point type in little-endian order from the underlying reader. Read more
source§

fn read_to_end<'a>(&'a mut self, buf: &'a mut Vec<u8>) -> ReadToEnd<'a, Self>
where Self: Unpin,

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

fn read_to_string<'a>( &'a mut self, dst: &'a mut String, ) -> ReadToString<'a, Self>
where Self: Unpin,

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

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

Creates an adaptor which reads at most limit bytes from it. Read more
source§

impl<T> Borrow<T> for T
where T: ?Sized,

source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
source§

impl<T> From<T> for T

source§

fn from(t: T) -> T

Returns the argument unchanged.

source§

impl<T, U> Into<U> for T
where U: From<T>,

source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

§

type Error = Infallible

The type returned in the event of a conversion error.
source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

Performs the conversion.
source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.