Added I/O impls for Pipe and halves.

Added `pair()`: Create 2 interlinked pipes for bi-directional communication.

Fixed transfering ownership of raw socket fds not surpressing destructor of previous owner type.

Fortune for comfork's current commit: Curse − 凶

src/lib.rs

@@ -2,6 +2,8 @@
 
 #[macro_use] extern crate cfg_if;
 
+use std::convert::{TryFrom, TryInto};
+
 #[macro_use] mod ext; use ext::*;
 mod sys;
 

src/sys/errno.rs

@@ -119,6 +119,15 @@ impl fmt::Display for Errno
     }
 }
 
+impl From<Errno> for std::io::Error
+{
+    #[inline] fn from(from: Errno) -> Self
+    {
+	std::io::Error::from_raw_os_error(from.errno.get())
+    }
+}
+
+
 remove! {
     /// Wrapping of errno around another Rust error type.
     #[derive(Debug)]

src/sys/pipe.rs

@@ -5,6 +5,7 @@ use super::*;
 use errno::Errno;
 use std::{error, fmt};
 use std::ops::Drop;
+use std::io;
 
 /// Kind of pipe error
 #[derive(Debug)]
@@ -54,6 +55,7 @@ pub struct Pipe
     rx: i32,
 }
 
+// Make sure when operating on consumers of Pipe of Read/WriteHalf, the pipe structures are `mem::forget`ed before returning, since the Drop impl of those types closes the raw socket.
 impl Pipe
 {
     /// Split this pipe into a read and write half.
@@ -61,7 +63,9 @@ impl Pipe
     /// Data written to the `WriteHalf` is sent to the `ReadHalf` (unless the pipe is mismatched.)
     #[inline] pub fn split(self) -> (WriteHalf, ReadHalf)
     {
-	(WriteHalf(self.tx), ReadHalf(self.rx))
+	let rv = (WriteHalf(self.tx), ReadHalf(self.rx));	
+	std::mem::forget(self);
+	rv
     }
 
     /// Create a `Pipe` from two halves.
@@ -71,7 +75,9 @@ impl Pipe
     /// If they are not, then writing to the `Pipe` will send the data to its original receiver, and reading from it will take the data from its original sender.
     #[inline] pub fn unsplit(tx: WriteHalf, rx: ReadHalf) -> Self
     {
-	Self::new_from_raw((tx.0, rx.0))
+	let v = Self::new_from_raw((tx.0, rx.0));
+	std::mem::forget((tx, rx));
+	v
     }
 
     /// Create a new `Pipe` from a pair of raw file descriptors.
@@ -88,14 +94,100 @@ impl Pipe
     /// # Safety
     /// The caller must check that `tx` and `rx` are valid, open file descriptors
     // TODO: When writing the async version make sure to state that the fds need to be non-blocking.
-    /// Does not check that the integers provided are valid and open file descriptors.
     #[inline] pub unsafe fn from_raw(tx: i32, rx: i32) -> Self
     {
 	Self::new_from_raw((tx, rx))
     }
+
+    /// Transfer the ownership of the socket's inner Tx and Rx fds.
+    pub fn into_raw(self) -> (i32, i32)
+    {
+	let txrx = (self.tx, self.rx);
+	std::mem::forget(self);
+	txrx
+    }
+
+    /// Create a new `Pipe` from two new file descriptors that stream to eachother.
+    ///
+    /// Data written to this pipe will also be read from it.
+    /// The pipe can be `split()`, and then `unsplit()` with a different split pipe's half to create a stream between those two pipes.
+    #[inline] pub fn new() -> Result<Self, Error>
+    {
+	let sk = unix_pipe()?;
+	Ok(Self::new_from_raw(sk))
+    }
+}
+
+/// Create a pair of bi-directional linked pipes.
+///
+/// Writing into one of the pipes will send the data to the other one, and vise-versa.
+pub fn pair() -> Result<(Pipe, Pipe), Error>
+{
+    let (a, b) = (Pipe::new()?, Pipe::new()?);
+    let (atx, arx) = a.split();
+    let (btx, brx) = b.split();
+    Ok((
+	Pipe::unsplit(atx, brx),
+	Pipe::unsplit(btx, arx),
+    ))
+}
+
+/// Write to a raw fd.
+fn write_raw(sock: i32, data: &[u8]) -> Result<usize, Errno>
+{
+    match unsafe {libc::write(sock, data.as_ptr() as *const _, data.len())} {
+	x if x < 0 => Err(Errno::last().unwrap_err()),
+	x => Ok(usize::try_from(x).unwrap()),
+    }
+}
+
+/// Read from a raw fd.
+fn read_raw(sock: i32, data: &mut [u8]) -> Result<usize, Errno>
+{   
+    match unsafe {libc::read(sock, data.as_ptr() as *mut _, data.len())} {
+	x if x < 0 => Err(Errno::last().unwrap_err()),
+	x => Ok(usize::try_from(x).unwrap()),
+    }
+}
+
+// There's nothing to flush, the fd isn't buffered.
+// fn flush_raw(sock: i32) -> Result<(), Errno>
+// {
+//     libc::fsync(sock)
+// }
+
+impl io::Read for Pipe
+{
+    #[inline] fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+	Ok(read_raw(self.rx, buf)?)
+    }
+}
+impl io::Write for Pipe
+{
+    #[inline] fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+	Ok(write_raw(self.tx, buf)?)
+    }
+    #[inline] fn flush(&mut self) -> io::Result<()> {
+	Ok(())
+    }
+}
+
+impl io::Read for ReadHalf
+{
+    #[inline] fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+	Ok(read_raw(self.0, buf)?)
+    }
+}
+impl io::Write for WriteHalf
+{
+    #[inline] fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+	Ok(write_raw(self.0, buf)?)
+    }
+    #[inline] fn flush(&mut self) -> io::Result<()> {
+	Ok(())
+    }
 }
 
-//TODO: Impl io::Read +/ io::Write for the types above
 
 impl Drop for ReadHalf
 {