impl Receiver

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.

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.

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};
# use std::error::Error;

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

# async fn dox() -> Result<(), Box<dyn Error>> {
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 */
}
# Ok(())
# }

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.

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.

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.

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(())
}

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.

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.

Notes

To avoid unnecessary syscalls, this will only attempt the read operation if the OS has informed Tokio that this pipe has become readable. Because of this, try_read() may fail with a WouldBlock error if Tokio has not yet heard from the OS that this pipe has become readable.

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(())
}

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.

Notes

To avoid unnecessary syscalls, this will only attempt the read operation if the OS has informed Tokio that this pipe has become readable. Because of this, try_read_vectored() may fail with a WouldBlock error if Tokio has not yet heard from the OS that this pipe has become readable.

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(())
}

fn try_io<R>(&self, f: impl FnOnce() -> Result<R>) -> Result<R>

Tries to read to the socket using a user-provided IO operation.

If the socket is ready, the provided closure is called. The closure should attempt to perform IO operation on the socket by manually calling the appropriate syscall. If the operation fails because the socket is not actually ready, then the closure should return a WouldBlock error and the readiness flag is cleared. The return value of the closure is then returned by try_io.

If the socket is not ready, then the closure is not called and a WouldBlock error is returned.

The closure should only return a WouldBlock error if it has performed an IO operation on the socket that failed due to the socket not being ready. Returning a WouldBlock error in any other situation will incorrectly clear the readiness flag, which can cause the socket to behave incorrectly.

The closure should not perform the IO operation using any of the methods defined on the Tokio pipe::Receiver type, as this will mess with the readiness flag and can cause the socket to behave incorrectly.

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

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.

Notes

To avoid unnecessary syscalls, this will only attempt the read operation if the OS has informed Tokio that this pipe has become readable. Because of this, try_read_buf() may fail with a WouldBlock error if Tokio has not yet heard from the OS that this pipe has become readable.

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(())
}

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.

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.