Function mio_pipe::new_pipe

pub fn new_pipe() -> Result<(Sender, Receiver)>

Create a new non-blocking Unix pipe.

This is a wrapper around Unix's pipe(2) system call and can be used as inter-process or thread communication channel.

This channel may be created before forking the process and then one end used in each process, e.g. the parent process has the sending end to send command to the child process.

Events

The Sender can be registered with WRITABLE interest to receive writable events, the Receiver with READABLE interest. Once data is written to the Sender the Receiver will receive an readable event.

In addition to those events, events will also be generated if the other side is dropped. However due to platform differences checking is_{read, write}_closed is not enough. To check if the Sender is dropped you'll need to check both is_error and is_read_closed on events for the Receiver, if either is true the Sender is dropped. On the Sender end check is_error and is_write_closed, if either is true the Receiver was dropped. Also see the second example below.

Deregistering

Both Sender and Receiver will deregister themselves when dropped, iff the file descriptors are not duplicated (via dup(2)).

Examples

Simple example that writes data into the sending end and read it from the receiving end.

use std::io::{self, Read, Write};

use mio::{Poll, Events, Interest, Token};
use mio_pipe::new_pipe;

// Unique tokens for the two ends of the channel.
const PIPE_RECV: Token = Token(0);
const PIPE_SEND: Token = Token(1);

// Create our `Poll` instance and the `Events` container.
let mut poll = Poll::new()?;
let mut events = Events::with_capacity(8);

// Create a new pipe.
let (mut sender, mut receiver) = new_pipe()?;

// Register both ends of the channel.
poll.registry().register(&mut receiver, PIPE_RECV, Interest::READABLE)?;
poll.registry().register(&mut sender, PIPE_SEND, Interest::WRITABLE)?;

const MSG: &[u8; 11] = b"Hello world";

loop {
    poll.poll(&mut events, None)?;

    for event in events.iter() {
        match event.token() {
            PIPE_SEND => sender.write(MSG)
                .and_then(|n| if n != MSG.len() {
                        // We'll consider a short write an error in this
                        // example. NOTE: we can't use `write_all` with
                        // non-blocking I/O.
                        Err(io::ErrorKind::WriteZero.into())
                    } else {
                        Ok(())
                    })?,
            PIPE_RECV => {
                let mut buf = [0; 11];
                let n = receiver.read(&mut buf)?;
                println!("received: {:?}", &buf[0..n]);
                assert_eq!(n, MSG.len());
                assert_eq!(&buf, &*MSG);
                return Ok(());
            },
            _ => unreachable!(),
        }
    }
}

Example that receives an event once the Sender is dropped.

// Same setup as in the example above.
let mut poll = Poll::new()?;
let mut events = Events::with_capacity(8);

let (mut sender, mut receiver) = new_pipe()?;

poll.registry().register(&mut receiver, PIPE_RECV, Interest::READABLE)?;
poll.registry().register(&mut sender, PIPE_SEND, Interest::WRITABLE)?;

// Drop the sender.
drop(sender);

poll.poll(&mut events, None)?;

for event in events.iter() {
    match event.token() {
        PIPE_RECV if event.is_error() || event.is_read_closed() => {
            // Detected that the sender was dropped.
            println!("Sender dropped!");
            return Ok(());
        },
        _ => unreachable!(),
    }
}