Add Stream and Sink future traits to IpcSender and Receiver

Cargo.toml

@@ -19,6 +19,7 @@ rand = "0.3"
 serde = "0.9"
 uuid = {version = "0.4", features = ["v4"]}
 fnv = "1.0.3"
+futures = "0.1"
 
 [target.'cfg(any(target_os = "linux", target_os = "freebsd"))'.dependencies]
 mio = "0.6.1"

src/ipc.rs

@@ -10,12 +10,13 @@
 use platform::{self, OsIpcChannel, OsIpcReceiver, OsIpcReceiverSet, OsIpcSender};
 use platform::{OsIpcOneShotServer, OsIpcSelectionResult, OsIpcSharedMemory, OsOpaqueIpcChannel};
 
+use futures::{Stream, Sink, Poll, Async, StartSend};
 use bincode;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use std::cell::RefCell;
 use std::cmp::min;
 use std::fmt::{self, Debug, Formatter};
-use std::io::Error;
+use std::io::{Error, ErrorKind};
 use std::marker::PhantomData;
 use std::mem;
 use std::ops::Deref;
@@ -86,6 +87,24 @@ impl<T> IpcReceiver<T> where T: Deserialize + Serialize {
     }
 }
 
+impl<T> Stream for IpcReceiver<T> where T: Deserialize + Serialize {
+    type Item = T;
+    type Error = bincode::Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        match self.try_recv() {
+            Ok(msg) => Ok(Async::Ready(Some(msg))),
+            Err(e) => match *e {
+                bincode::ErrorKind::IoError(ref e) if e.kind() == ErrorKind::ConnectionReset =>
+                    Ok(Async::Ready(None)),
+                bincode::ErrorKind::IoError(ref e) if e.kind() == ErrorKind::WouldBlock =>
+                    Ok(Async::NotReady),
+                _ => Err(e),
+            },
+        }
+    }
+}
+
 impl<T> Deserialize for IpcReceiver<T> where T: Deserialize + Serialize {
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error> where D: Deserializer {
         let index: usize = try!(Deserialize::deserialize(deserializer));
@@ -173,6 +192,19 @@ impl<T> IpcSender<T> where T: Serialize {
     }
 }
 
+impl<T> Sink for IpcSender<T> where T: Serialize {
+    type SinkItem = T;
+    type SinkError = bincode::Error;
+
+    fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
+        unimplemented!();
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        unimplemented!();
+    }
+}
+
 impl<T> Deserialize for IpcSender<T> where T: Serialize {
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error> where D: Deserializer {
         let os_sender = try!(deserialize_os_ipc_sender(deserializer));

src/lib.rs

@@ -30,6 +30,7 @@ extern crate fnv;
           target_os="linux"))]
 #[macro_use]
 extern crate syscall;
+extern crate futures;
 
 
 pub mod ipc;