Upgrade tokio_core dependency to use tokio 0.1

tokio-rs/tokio-rfcs#2

Author: Markus Westerlind

.travis.yml

@@ -18,6 +18,7 @@ matrix:
 
 script:
   - cargo test
+  - cargo test --features unstable-futures
 
 env:
   global:

Cargo.toml

@@ -1,27 +1,31 @@
-[package]
-name = "tokio-uds"
-version = "0.1.7"
-authors = ["Alex Crichton <alex@alexcrichton.com>"]
-license = "MIT/Apache-2.0"
-repository = "https://github.com/tokio-rs/tokio-uds"
-homepage = "https://github.com/tokio-rs/tokio-uds"
-documentation = "https://docs.rs/tokio-uds"
-description = """
-Unix Domain sockets for Tokio
-"""
-categories = ["asynchronous"]
-
-[badges]
-travis-ci = { repository = "tokio-rs/tokio-uds" }
-appveyor = { repository = "alexcrichton/tokio-uds" }
-
-[dependencies]
-bytes = "0.4"
-futures = "0.1.11"
-iovec = "0.1"
-libc = "0.2"
-log = "0.4"
-mio = "0.6.5"
-mio-uds = "0.6.4"
-tokio-core = "0.1"
-tokio-io = "0.1"
+[package]
+name = "tokio-uds"
+version = "0.1.7"
+authors = ["Alex Crichton <alex@alexcrichton.com>"]
+license = "MIT/Apache-2.0"
+repository = "https://github.com/tokio-rs/tokio-uds"
+homepage = "https://github.com/tokio-rs/tokio-uds"
+documentation = "https://docs.rs/tokio-uds"
+description = """
+Unix Domain sockets for Tokio
+"""
+categories = ["asynchronous"]
+
+[badges]
+travis-ci = { repository = "tokio-rs/tokio-uds" }
+appveyor = { repository = "alexcrichton/tokio-uds" }
+
+[dependencies]
+bytes = "0.4"
+futures = "0.1"
+futures2 = { version = "0.1.0", optional = true }
+iovec = "0.1"
+libc = "0.2"
+log = "0.4"
+mio = "0.6.14"
+mio-uds = "0.6.4"
+tokio-reactor = { git = "https://github.com/tokio-rs/tokio", features = ["unstable-futures"] }
+tokio-io = "0.1"
+
+[features]
+unstable-futures = ["futures2"]

src/frame.rs

@@ -2,7 +2,10 @@ use std::io;
 use std::os::unix::net::SocketAddr;
 use std::path::PathBuf;
 
-use futures::{Async, Poll, Stream, Sink, StartSend, AsyncSink};
+use futures::{Async, AsyncSink, Poll, Sink, StartSend, Stream};
+
+#[cfg(feature = "unstable-futures")]
+use futures2::{self, task};
 
 use UnixDatagram;
 
@@ -50,8 +53,7 @@ pub trait UnixDatagramCodec {
     ///
     /// The encode method also determines the destination to which the buffer
     /// should be directed, which will be returned as a `SocketAddr`.
-    fn encode(&mut self, msg: Self::Out, buf: &mut Vec<u8>)
-              -> io::Result<PathBuf>;
+    fn encode(&mut self, msg: Self::Out, buf: &mut Vec<u8>) -> io::Result<PathBuf>;
 }
 
 /// A unified `Stream` and `Sink` interface to an underlying
@@ -73,14 +75,28 @@ impl<C: UnixDatagramCodec> Stream for UnixDatagramFramed<C> {
     type Error = io::Error;
 
     fn poll(&mut self) -> Poll<Option<C::In>, io::Error> {
-        let (n, addr) = try_nb!(self.socket.recv_from(&mut self.rd));
+        let (n, addr) = try_ready!(self.socket.recv_from(&mut self.rd));
         trace!("received {} bytes, decoding", n);
         let frame = try!(self.codec.decode(&addr, &self.rd[..n]));
         trace!("frame decoded from buffer");
         Ok(Async::Ready(Some(frame)))
     }
 }
 
+#[cfg(feature = "unstable-futures")]
+impl<C: UnixDatagramCodec> futures2::Stream for UnixDatagramFramed<C> {
+    type Item = C::In;
+    type Error = io::Error;
+
+    fn poll_next(&mut self, cx: &mut task::Context) -> futures2::Poll<Option<C::In>, io::Error> {
+        let (n, addr) = try_ready2!(self.socket.recv_from2(cx, &mut self.rd));
+        trace!("received {} bytes, decoding", n);
+        let frame = try!(self.codec.decode(&addr, &self.rd[..n]));
+        trace!("frame decoded from buffer");
+        Ok(futures2::Async::Ready(Some(frame)))
+    }
+}
+
 impl<C: UnixDatagramCodec> Sink for UnixDatagramFramed<C> {
     type SinkItem = C::Out;
     type SinkError = io::Error;
@@ -101,19 +117,21 @@ impl<C: UnixDatagramCodec> Sink for UnixDatagramFramed<C> {
         trace!("flushing framed transport");
 
         if self.wr.is_empty() {
-            return Ok(Async::Ready(()))
+            return Ok(Async::Ready(()));
         }
 
         trace!("writing; remaining={}", self.wr.len());
-        let n = try_nb!(self.socket.send_to(&self.wr, &self.out_addr));
+        let n = try_ready!(self.socket.send_to(&self.wr, &self.out_addr));
         trace!("written {}", n);
         let wrote_all = n == self.wr.len();
         self.wr.clear();
         if wrote_all {
             Ok(Async::Ready(()))
         } else {
-            Err(io::Error::new(io::ErrorKind::Other,
-                               "failed to write entire datagram to socket"))
+            Err(io::Error::new(
+                io::ErrorKind::Other,
+                "failed to write entire datagram to socket",
+            ))
         }
     }
 
@@ -123,6 +141,53 @@ impl<C: UnixDatagramCodec> Sink for UnixDatagramFramed<C> {
     }
 }
 
+#[cfg(feature = "unstable-futures")]
+impl<C: UnixDatagramCodec> futures2::Sink for UnixDatagramFramed<C> {
+    type SinkItem = C::Out;
+    type SinkError = io::Error;
+
+    fn poll_ready(&mut self, cx: &mut task::Context) -> futures2::Poll<(), io::Error> {
+        if self.wr.len() > 0 {
+            try!(self.poll_flush(cx));
+            if self.wr.len() > 0 {
+                return Ok(futures2::Async::Pending);
+            }
+        }
+        Ok(().into())
+    }
+
+    fn start_send(&mut self, item: C::Out) -> Result<(), io::Error> {
+        self.out_addr = try!(self.codec.encode(item, &mut self.wr));
+        Ok(())
+    }
+
+    fn poll_flush(&mut self, cx: &mut task::Context) -> futures2::Poll<(), io::Error> {
+        trace!("flushing framed transport");
+
+        if self.wr.is_empty() {
+            return Ok(futures2::Async::Ready(()));
+        }
+
+        trace!("writing; remaining={}", self.wr.len());
+        let n = try_ready2!(self.socket.send_to2(cx, &self.wr, &self.out_addr));
+        trace!("written {}", n);
+        let wrote_all = n == self.wr.len();
+        self.wr.clear();
+        if wrote_all {
+            Ok(futures2::Async::Ready(()))
+        } else {
+            Err(io::Error::new(
+                io::ErrorKind::Other,
+                "failed to write entire datagram to socket",
+            ))
+        }
+    }
+
+    fn poll_close(&mut self, cx: &mut task::Context) -> futures2::Poll<(), io::Error> {
+        self.poll_flush(cx)
+    }
+}
+
 pub fn new<C: UnixDatagramCodec>(socket: UnixDatagram, codec: C) -> UnixDatagramFramed<C> {
     UnixDatagramFramed {
         socket: socket,