Migrate to using tokio-io

Deprecate the existing `io` module in this crate entirely.

More details coming soon!

Closes #61

src/channel.rs

@@ -5,6 +5,7 @@
 
 #![deprecated(since = "0.1.1", note = "use `futures::sync::mpsc` instead")]
 #![allow(deprecated)]
+#![cfg(feature = "with-deprecated")]
 
 use std::io;
 use std::sync::mpsc::TryRecvError;
@@ -95,6 +96,10 @@ impl<T> Sink for Sender<T> {
     fn poll_complete(&mut self) -> Poll<(), io::Error> {
         Ok(().into())
     }
+
+    fn close(&mut self) -> Poll<(), io::Error> {
+        Ok(().into())
+    }
 }
 
 impl<T> Clone for Sender<T> {

src/io/frame.rs

@@ -404,6 +404,11 @@ impl<T: Io, C: Codec> Sink for Framed<T, C> {
         trace!("framed transport flushed");
         return Ok(Async::Ready(()));
     }
+
+    fn close(&mut self) -> Poll<(), io::Error> {
+        try_ready!(self.poll_complete());
+        Ok(().into())
+    }
 }
 
 pub fn framed<T, C>(io: T, codec: C) -> Framed<T, C> {

src/io/mod.rs

@@ -9,12 +9,14 @@
 //! [found online]: https://tokio.rs/docs/getting-started/core/
 //! [low level details]: https://tokio.rs/docs/going-deeper/core-low-level/
 
+#![deprecated(note = "moved to the `tokio-io` crate")]
+
 use std::io;
 
-use futures::{BoxFuture, Async, Poll};
+use futures::{Async, Poll};
+use futures::future::BoxFuture;
 use futures::stream::BoxStream;
-
-use mio::IoVec;
+use iovec::IoVec;
 
 /// A convenience typedef around a `Future` whose error component is `io::Error`
 pub type IoFuture<T> = BoxFuture<T, io::Error>;
@@ -138,7 +140,7 @@ pub trait Io: io::Read + io::Write {
         if bufs.is_empty() {
             Ok(0)
         } else {
-            self.read(bufs[0].as_mut_bytes())
+            self.read(&mut bufs[0])
         }
     }
 
@@ -163,7 +165,7 @@ pub trait Io: io::Read + io::Write {
         if bufs.is_empty() {
             Ok(0)
         } else {
-            self.write(bufs[0].as_bytes())
+            self.write(&bufs[0])
         }
     }
 

src/io/split.rs

@@ -2,7 +2,6 @@ use std::io::{self, Read, Write};
 
 use futures::Async;
 use futures::sync::BiLock;
-use mio;
 
 use io::Io;
 
@@ -47,7 +46,7 @@ impl<T: Read> Read for ReadHalf<T> {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         match self.handle.poll_lock() {
             Async::Ready(mut l) => l.read(buf),
-            Async::NotReady => Err(mio::would_block()),
+            Async::NotReady => Err(::would_block()),
         }
     }
 }
@@ -56,14 +55,14 @@ impl<T: Write> Write for WriteHalf<T> {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
         match self.handle.poll_lock() {
             Async::Ready(mut l) => l.write(buf),
-            Async::NotReady => Err(mio::would_block()),
+            Async::NotReady => Err(::would_block()),
         }
     }
 
     fn flush(&mut self) -> io::Result<()> {
         match self.handle.poll_lock() {
             Async::Ready(mut l) => l.flush(),
-            Async::NotReady => Err(mio::would_block()),
+            Async::NotReady => Err(::would_block()),
         }
     }
 }

src/lib.rs

@@ -90,9 +90,13 @@
 #![doc(html_root_url = "https://docs.rs/tokio-core/0.1")]
 #![deny(missing_docs)]
 
+extern crate bytes;
+#[macro_use]
 extern crate futures;
+extern crate iovec;
 extern crate mio;
 extern crate slab;
+extern crate tokio_io;
 
 #[macro_use]
 extern crate scoped_tls;
@@ -108,3 +112,9 @@ mod heap;
 pub mod channel;
 pub mod net;
 pub mod reactor;
+
+use std::io as sio;
+
+fn would_block() -> sio::Error {
+    sio::Error::new(sio::ErrorKind::WouldBlock, "would block")
+}

src/net/tcp.rs

@@ -3,12 +3,14 @@ use std::io::{self, Read, Write};
 use std::mem;
 use std::net::{self, SocketAddr, Shutdown};
 
+use bytes::{Buf, BufMut};
 use futures::stream::Stream;
 use futures::sync::oneshot;
 use futures::{Future, Poll, Async};
+use iovec::IoVec;
 use mio;
+use tokio_io::{AsyncRead, AsyncWrite};
 
-use io::{Io, IoFuture};
 use reactor::{Handle, PollEvented};
 
 /// An I/O object representing a TCP socket listening for incoming connections.
@@ -60,7 +62,7 @@ impl TcpListener {
                 match pending.poll().expect("shouldn't be canceled") {
                     Async::NotReady => {
                         self.pending_accept = Some(pending);
-                        return Err(mio::would_block())
+                        return Err(::would_block())
                     },
                     Async::Ready(r) => return r,
                 }
@@ -94,7 +96,7 @@ impl TcpListener {
                             .map(move |io| {
                                 (TcpStream { io: io }, addr)
                             });
-                        tx.complete(res);
+                        drop(tx.send(res));
                         Ok(())
                     });
                     self.pending_accept = Some(rx);
@@ -299,7 +301,8 @@ impl TcpStream {
     ///   (perhaps to `INADDR_ANY`) before this method is called.
     pub fn connect_stream(stream: net::TcpStream,
                           addr: &SocketAddr,
-                          handle: &Handle) -> IoFuture<TcpStream> {
+                          handle: &Handle)
+                          -> Box<Future<Item=TcpStream, Error=io::Error> + Send> {
         let state = match mio::tcp::TcpStream::connect_stream(stream, addr) {
             Ok(tcp) => TcpStream::new(tcp, handle),
             Err(e) => TcpStreamNewState::Error(e),
@@ -425,7 +428,28 @@ impl Write for TcpStream {
     }
 }
 
-impl Io for TcpStream {
+impl AsyncRead for TcpStream {
+    unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool {
+        false
+    }
+
+    fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> Poll<usize, io::Error> {
+        <&TcpStream>::read_buf(&mut &*self, buf)
+    }
+}
+
+impl AsyncWrite for TcpStream {
+    fn shutdown(&mut self) -> Poll<(), io::Error> {
+        <&TcpStream>::shutdown(&mut &*self)
+    }
+
+    fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Poll<usize, io::Error> {
+        <&TcpStream>::write_buf(&mut &*self, buf)
+    }
+}
+
+#[allow(deprecated)]
+impl ::io::Io for TcpStream {
     fn poll_read(&mut self) -> Async<()> {
         <TcpStream>::poll_read(self)
     }
@@ -434,29 +458,26 @@ impl Io for TcpStream {
         <TcpStream>::poll_write(self)
     }
 
-    fn read_vec(&mut self, bufs: &mut [&mut mio::IoVec]) -> io::Result<usize> {
-        if let Async::NotReady = self.poll_read() {
-            return Err(mio::would_block())
+    fn read_vec(&mut self, bufs: &mut [&mut IoVec]) -> io::Result<usize> {
+        if let Async::NotReady = <TcpStream>::poll_read(self) {
+            return Err(::would_block())
         }
         let r = self.io.get_ref().read_bufs(bufs);
         if is_wouldblock(&r) {
             self.io.need_read();
         }
         return r
-
-
     }
 
-    fn write_vec(&mut self, bufs: &[&mio::IoVec]) -> io::Result<usize> {
-        if let Async::NotReady = self.poll_write() {
-            return Err(mio::would_block())
+    fn write_vec(&mut self, bufs: &[&IoVec]) -> io::Result<usize> {
+        if let Async::NotReady = <TcpStream>::poll_write(self) {
+            return Err(::would_block())
         }
         let r = self.io.get_ref().write_bufs(bufs);
         if is_wouldblock(&r) {
             self.io.need_write();
         }
         return r
-
     }
 }
 
@@ -483,7 +504,62 @@ impl<'a> Write for &'a TcpStream {
     }
 }
 
-impl<'a> Io for &'a TcpStream {
+impl<'a> AsyncRead for &'a TcpStream {
+    unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool {
+        false
+    }
+
+    fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> Poll<usize, io::Error> {
+        if let Async::NotReady = <TcpStream>::poll_read(self) {
+            return Err(::would_block())
+        }
+        let mut bufs: [_; 16] = Default::default();
+        unsafe {
+            let n = buf.bytes_vec_mut(&mut bufs);
+            match self.io.get_ref().read_bufs(&mut bufs[..n]) {
+                Ok(n) => {
+                    buf.advance_mut(n);
+                    Ok(Async::Ready(n))
+                }
+                Err(e) => {
+                    if e.kind() == io::ErrorKind::WouldBlock {
+                        self.io.need_write();
+                    }
+                    Err(e)
+                }
+            }
+        }
+    }
+}
+
+impl<'a> AsyncWrite for &'a TcpStream {
+    fn shutdown(&mut self) -> Poll<(), io::Error> {
+        Ok(().into())
+    }
+
+    fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Poll<usize, io::Error> {
+        if let Async::NotReady = <TcpStream>::poll_write(self) {
+            return Err(::would_block())
+        }
+        let mut bufs: [_; 16] = Default::default();
+        let n = buf.bytes_vec(&mut bufs);
+        match self.io.get_ref().write_bufs(&bufs[..n]) {
+            Ok(n) => {
+                buf.advance(n);
+                Ok(Async::Ready(n))
+            }
+            Err(e) => {
+                if e.kind() == io::ErrorKind::WouldBlock {
+                    self.io.need_write();
+                }
+                Err(e)
+            }
+        }
+    }
+}
+
+#[allow(deprecated)]
+impl<'a> ::io::Io for &'a TcpStream {
     fn poll_read(&mut self) -> Async<()> {
         <TcpStream>::poll_read(self)
     }

src/net/udp/frame.rs

@@ -111,6 +111,11 @@ impl<C: UdpCodec> Sink for UdpFramed<C> {
                                "failed to write entire datagram to socket"))
         }
     }
+
+    fn close(&mut self) -> Poll<(), io::Error> {
+        try_ready!(self.poll_complete());
+        Ok(().into())
+    }
 }
 
 pub fn new<C: UdpCodec>(socket: UdpSocket, codec: C) -> UdpFramed<C> {

src/net/udp/mod.rs

@@ -101,13 +101,13 @@ impl UdpSocket {
     /// documentation for concrete examples.
     pub fn send_to(&self, buf: &[u8], target: &SocketAddr) -> io::Result<usize> {
         if let Async::NotReady = self.io.poll_write() {
-            return Err(mio::would_block())
+            return Err(::would_block())
         }
         match self.io.get_ref().send_to(buf, target) {
             Ok(Some(n)) => Ok(n),
             Ok(None) => {
                 self.io.need_write();
-                Err(mio::would_block())
+                Err(::would_block())
             }
             Err(e) => Err(e),
         }
@@ -144,13 +144,13 @@ impl UdpSocket {
     /// read and the address from whence the data came.
     pub fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
         if let Async::NotReady = self.io.poll_read() {
-            return Err(mio::would_block())
+            return Err(::would_block())
         }
         match self.io.get_ref().recv_from(buf) {
             Ok(Some(n)) => Ok(n),
             Ok(None) => {
                 self.io.need_read();
-                Err(mio::would_block())
+                Err(::would_block())
             }
             Err(e) => Err(e),
         }

src/reactor/io_token.rs

@@ -3,7 +3,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::io;
 
 use futures::task;
-use mio;
+use mio::event::Evented;
 
 use reactor::{Message, Remote, Handle, Direction};
 
@@ -31,7 +31,7 @@ impl IoToken {
     /// The returned future will panic if the event loop this handle is
     /// associated with has gone away, or if there is an error communicating
     /// with the event loop.
-    pub fn new(source: &mio::Evented, handle: &Handle) -> io::Result<IoToken> {
+    pub fn new(source: &Evented, handle: &Handle) -> io::Result<IoToken> {
         match handle.inner.upgrade() {
             Some(inner) => {
                 let (ready, token) = try!(inner.borrow_mut().add_source(source));

src/reactor/mod.rs

@@ -14,11 +14,13 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT, Ordering};
 use std::time::{Instant, Duration};
 
-use futures::{self, Future, IntoFuture, Async};
+use futures::{Future, IntoFuture, Async};
+use futures::future;
 use futures::executor::{self, Spawn, Unpark};
 use futures::sync::mpsc;
 use futures::task::Task;
 use mio;
+use mio::event::Evented;
 use slab::Slab;
 
 use heap::{Heap, Slot};
@@ -153,15 +155,17 @@ impl Core {
     /// creation.
     pub fn new() -> io::Result<Core> {
         let io = try!(mio::Poll::new());
-        let future_pair = mio::Registration::new(&io,
-                                                 TOKEN_FUTURE,
-                                                 mio::Ready::readable(),
-                                                 mio::PollOpt::level());
+        let future_pair = mio::Registration::new2();
+        try!(io.register(&future_pair.0,
+                         TOKEN_FUTURE,
+                         mio::Ready::readable(),
+                         mio::PollOpt::level()));
         let (tx, rx) = mpsc::unbounded();
-        let channel_pair = mio::Registration::new(&io,
-                                                  TOKEN_MESSAGES,
-                                                  mio::Ready::readable(),
-                                                  mio::PollOpt::level());
+        let channel_pair = mio::Registration::new2();
+        try!(io.register(&channel_pair.0,
+                         TOKEN_MESSAGES,
+                         mio::Ready::readable(),
+                         mio::PollOpt::level()));
         let rx_readiness = Arc::new(MySetReadiness(channel_pair.1));
         rx_readiness.unpark();
 
@@ -296,16 +300,16 @@ impl Core {
         for i in 0..self.events.len() {
             let event = self.events.get(i).unwrap();
             let token = event.token();
-            trace!("event {:?} {:?}", event.kind(), event.token());
+            trace!("event {:?} {:?}", event.readiness(), event.token());
 
             if token == TOKEN_MESSAGES {
-                self.rx_readiness.0.set_readiness(mio::Ready::none()).unwrap();
+                self.rx_readiness.0.set_readiness(mio::Ready::empty()).unwrap();
                 CURRENT_LOOP.set(&self, || self.consume_queue());
             } else if token == TOKEN_FUTURE {
-                self.future_readiness.0.set_readiness(mio::Ready::none()).unwrap();
+                self.future_readiness.0.set_readiness(mio::Ready::empty()).unwrap();
                 fired = true;
             } else {
-                self.dispatch(token, event.kind());
+                self.dispatch(token, event.readiness());
             }
         }
         debug!("loop process - {} events, {:?}", amt, after_poll.elapsed());
@@ -326,7 +330,7 @@ impl Core {
         let mut writer = None;
         let mut inner = self.inner.borrow_mut();
         if let Some(io) = inner.io_dispatch.get_mut(token) {
-            if ready.is_readable() || ready.is_hup() {
+            if ready.is_readable() || platform::is_hup(&ready) {
                 reader = io.reader.take();
                 io.readiness.fetch_or(Readiness::Readable as usize,
                     Ordering::Relaxed);
@@ -353,7 +357,7 @@ impl Core {
             Some(slot) => (slot.spawn.take(), slot.wake.clone()),
             None => return,
         };
-        wake.0.set_readiness(mio::Ready::none()).unwrap();
+        wake.0.set_readiness(mio::Ready::empty()).unwrap();
         let mut task = match task {
             Some(task) => task,
             None => return,
@@ -455,7 +459,7 @@ impl fmt::Debug for Core {
 }
 
 impl Inner {
-    fn add_source(&mut self, source: &mio::Evented)
+    fn add_source(&mut self, source: &Evented)
                   -> io::Result<(Arc<AtomicUsize>, usize)> {
         debug!("adding a new I/O source");
         let sched = ScheduledIo {
@@ -470,12 +474,14 @@ impl Inner {
         let entry = self.io_dispatch.vacant_entry().unwrap();
         try!(self.io.register(source,
                               mio::Token(TOKEN_START + entry.index() * 2),
-                              mio::Ready::readable() | mio::Ready::writable() | mio::Ready::hup(),
+                              mio::Ready::readable() |
+                                mio::Ready::writable() |
+                                platform::hup(),
                               mio::PollOpt::edge()));
         Ok((sched.readiness.clone(), entry.insert(sched).index()))
     }
 
-    fn deregister_source(&mut self, source: &mio::Evented) -> io::Result<()> {
+    fn deregister_source(&mut self, source: &Evented) -> io::Result<()> {
         self.io.deregister(source)
     }
 
@@ -546,10 +552,12 @@ impl Inner {
         }
         let entry = self.task_dispatch.vacant_entry().unwrap();
         let token = TOKEN_START + 2 * entry.index() + 1;
-        let pair = mio::Registration::new(&self.io,
-                                          mio::Token(token),
-                                          mio::Ready::readable(),
-                                          mio::PollOpt::level());
+        let pair = mio::Registration::new2();
+        self.io.register(&pair.0,
+                         mio::Token(token),
+                         mio::Ready::readable(),
+                         mio::PollOpt::level())
+            .expect("cannot fail future registration with mio");
         let unpark = Arc::new(MySetReadiness(pair.1));
         let entry = entry.insert(ScheduledTask {
             spawn: Some(executor::spawn(future)),
@@ -688,7 +696,7 @@ impl Handle {
         where F: FnOnce() -> R + 'static,
               R: IntoFuture<Item=(), Error=()> + 'static,
     {
-        self.spawn(futures::lazy(f))
+        self.spawn(future::lazy(f))
     }
 
     /// Return the ID of the represented Core
@@ -742,3 +750,30 @@ impl<F: FnOnce(&Core) + Send + 'static> FnBox for F {
         (*self)(lp)
     }
 }
+
+#[cfg(unix)]
+mod platform {
+    use mio::Ready;
+    use mio::unix::UnixReady;
+
+    pub fn is_hup(event: &Ready) -> bool {
+        UnixReady::from(*event).is_hup()
+    }
+
+    pub fn hup() -> Ready {
+        UnixReady::hup().into()
+    }
+}
+
+#[cfg(windows)]
+mod platform {
+    use mio::Ready;
+
+    pub fn is_hup(_event: &Ready) -> bool {
+        false
+    }
+
+    pub fn hup() -> Ready {
+        Ready::empty()
+    }
+}

src/reactor/poll_evented.rs

@@ -10,10 +10,10 @@ use std::fmt;
 use std::io::{self, Read, Write};
 use std::sync::atomic::{AtomicUsize, Ordering};
 
-use futures::Async;
-use mio;
+use futures::{Async, Poll};
+use mio::event::Evented;
+use tokio_io::{AsyncRead, AsyncWrite};
 
-use io::Io;
 use reactor::{Handle, Remote};
 use reactor::Readiness::*;
 use reactor::io_token::IoToken;
@@ -45,7 +45,7 @@ pub struct PollEvented<E> {
     io: E,
 }
 
-impl<E: mio::Evented + fmt::Debug> fmt::Debug for PollEvented<E> {
+impl<E: Evented + fmt::Debug> fmt::Debug for PollEvented<E> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         f.debug_struct("PollEvented")
          .field("io", &self.io)
@@ -53,7 +53,7 @@ impl<E: mio::Evented + fmt::Debug> fmt::Debug for PollEvented<E> {
     }
 }
 
-impl<E: mio::Evented> PollEvented<E> {
+impl<E: Evented> PollEvented<E> {
     /// Creates a new readiness stream associated with the provided
     /// `loop_handle` and for the given `source`.
     ///
@@ -193,7 +193,7 @@ impl<E> PollEvented<E> {
 impl<E: Read> Read for PollEvented<E> {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         if let Async::NotReady = self.poll_read() {
-            return Err(mio::would_block())
+            return Err(::would_block())
         }
         let r = self.get_mut().read(buf);
         if is_wouldblock(&r) {
@@ -206,7 +206,7 @@ impl<E: Read> Read for PollEvented<E> {
 impl<E: Write> Write for PollEvented<E> {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
         if let Async::NotReady = self.poll_write() {
-            return Err(mio::would_block())
+            return Err(::would_block())
         }
         let r = self.get_mut().write(buf);
         if is_wouldblock(&r) {
@@ -217,7 +217,7 @@ impl<E: Write> Write for PollEvented<E> {
 
     fn flush(&mut self) -> io::Result<()> {
         if let Async::NotReady = self.poll_write() {
-            return Err(mio::would_block())
+            return Err(::would_block())
         }
         let r = self.get_mut().flush();
         if is_wouldblock(&r) {
@@ -227,7 +227,17 @@ impl<E: Write> Write for PollEvented<E> {
     }
 }
 
-impl<E: Read + Write> Io for PollEvented<E> {
+impl<E: Read> AsyncRead for PollEvented<E> {
+}
+
+impl<E: Write> AsyncWrite for PollEvented<E> {
+    fn shutdown(&mut self) -> Poll<(), io::Error> {
+        Ok(().into())
+    }
+}
+
+#[allow(deprecated)]
+impl<E: Read + Write> ::io::Io for PollEvented<E> {
     fn poll_read(&mut self) -> Async<()> {
         <PollEvented<E>>::poll_read(self)
     }
@@ -242,7 +252,7 @@ impl<'a, E> Read for &'a PollEvented<E>
 {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         if let Async::NotReady = self.poll_read() {
-            return Err(mio::would_block())
+            return Err(::would_block())
         }
         let r = self.get_ref().read(buf);
         if is_wouldblock(&r) {
@@ -257,7 +267,7 @@ impl<'a, E> Write for &'a PollEvented<E>
 {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
         if let Async::NotReady = self.poll_write() {
-            return Err(mio::would_block())
+            return Err(::would_block())
         }
         let r = self.get_ref().write(buf);
         if is_wouldblock(&r) {
@@ -268,7 +278,7 @@ impl<'a, E> Write for &'a PollEvented<E>
 
     fn flush(&mut self) -> io::Result<()> {
         if let Async::NotReady = self.poll_write() {
-            return Err(mio::would_block())
+            return Err(::would_block())
         }
         let r = self.get_ref().flush();
         if is_wouldblock(&r) {
@@ -278,7 +288,21 @@ impl<'a, E> Write for &'a PollEvented<E>
     }
 }
 
-impl<'a, E> Io for &'a PollEvented<E>
+impl<'a, E> AsyncRead for &'a PollEvented<E>
+    where &'a E: Read,
+{
+}
+
+impl<'a, E> AsyncWrite for &'a PollEvented<E>
+    where &'a E: Write,
+{
+    fn shutdown(&mut self) -> Poll<(), io::Error> {
+        Ok(().into())
+    }
+}
+
+#[allow(deprecated)]
+impl<'a, E> ::io::Io for &'a PollEvented<E>
     where &'a E: Read + Write,
 {
     fn poll_read(&mut self) -> Async<()> {