Make IpcReceiver a futures::Stream

Extracts the IpcReceiver half of servo#165.

This is placed behind an `async` feature so the dependency on futures isn't
forced.

Author: spinda

Cargo.toml

@@ -10,6 +10,7 @@ repository = "https://github.com/servo/ipc-channel"
 force-inprocess = []
 memfd = ["syscall"]
 unstable = []
+async = ["futures"]
 
 [dependencies]
 bincode = "0.8"
@@ -24,6 +25,7 @@ fnv = "1.0.3"
 mio = "0.6.1"
 
 syscall = { version = "0.2.1", optional = true }
+futures = { version = "0.1", optional = true }
 
 [dev-dependencies]
 crossbeam = "0.2"

src/ipc.rs

@@ -20,6 +20,11 @@ use std::marker::PhantomData;
 use std::mem;
 use std::ops::Deref;
 
+#[cfg(feature = "async")]
+use futures::{Async, Poll, Stream};
+#[cfg(feature = "async")]
+use std::io::ErrorKind;
+
 thread_local! {
     static OS_IPC_CHANNELS_FOR_DESERIALIZATION: RefCell<Vec<OsOpaqueIpcChannel>> =
         RefCell::new(Vec::new())
@@ -86,6 +91,27 @@ impl<T> IpcReceiver<T> where T: for<'de> Deserialize<'de> + Serialize {
     }
 }
 
+#[cfg(feature = "async")]
+impl<T> Stream for IpcReceiver<T> where T: for<'de> Deserialize<'de> + 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(Some(msg).into()),
+            Err(err) => match *err {
+                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(err),
+            },
+        }
+    }
+}
+
 impl<'de, T> Deserialize<'de> for IpcReceiver<T> where T: for<'dde> Deserialize<'dde> + Serialize {
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error> where D: Deserializer<'de> {
         let index: usize = try!(Deserialize::deserialize(deserializer));

src/lib.rs

@@ -31,6 +31,9 @@ extern crate fnv;
 #[macro_use]
 extern crate syscall;
 
+#[cfg(feature = "async")]
+extern crate futures;
+
 
 pub mod ipc;
 pub mod platform;