Update futures-minihttp

src/bin/echo.rs

@@ -5,14 +5,11 @@ extern crate futures_io;
 extern crate futures_mio;
 
 use std::env;
-use std::io::{self, Read, Write};
 use std::net::SocketAddr;
-use std::sync::Arc;
 
 use futures::Future;
 use futures::stream::Stream;
-use futures_io::copy;
-use futures_mio::TcpStream;
+use futures_io::{copy, TaskIo};
 
 fn main() {
     let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
@@ -36,8 +33,9 @@ fn main() {
         // Finally we use the `io::copy` future to copy all data from the
         // reading half onto the writing half.
         socket.incoming().for_each(|(socket, addr)| {
-            let socket = Arc::new(socket);
-            let amt = copy(SocketIo(socket.clone()), SocketIo(socket));
+            let socket = futures::lazy(|| futures::finished(TaskIo::new(socket)));
+            let pair = socket.map(|s| s.split());
+            let amt = pair.and_then(|(reader, writer)| copy(reader, writer));
 
             // Once all that is done we print out how much we wrote, and then
             // critically we *forget* this future which allows it to run
@@ -51,21 +49,3 @@ fn main() {
     });
     l.run(done).unwrap();
 }
-
-struct SocketIo(Arc<TcpStream>);
-
-impl Read for SocketIo {
-    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        (&*self.0).read(buf)
-    }
-}
-
-impl Write for SocketIo {
-    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        (&*self.0).write(buf)
-    }
-
-    fn flush(&mut self) -> io::Result<()> {
-        (&*self.0).flush()
-    }
-}

src/event_loop.rs

@@ -995,8 +995,10 @@ impl<T, U> LoopFuture<T, U>
                     lp.map(|lp| f(lp, data.take().unwrap()))
                 });
                 if let Some(ret) = ret {
+                    debug!("loop future done immediately on event loop");
                     return ret.into()
                 }
+                debug!("loop future needs to send info to event loop");
 
                 let task = task::park();
                 let result = Arc::new(Slot::new(None));

src/tcp.rs

@@ -100,8 +100,12 @@ impl TcpListener {
 
             fn poll(&mut self) -> Poll<Option<Self::Item>, io::Error> {
                 match self.inner.listener.io().accept() {
-                    Ok(Some(pair)) => Poll::Ok(Some(pair)),
+                    Ok(Some(pair)) => {
+                        debug!("accepted a socket");
+                        Poll::Ok(Some(pair))
+                    }
                     Ok(None) => {
+                        debug!("waiting to accept another socket");
                         self.inner.ready.need_read();
                         Poll::NotReady
                     }

tests/echo.rs

@@ -3,14 +3,13 @@ extern crate futures;
 extern crate futures_io;
 extern crate futures_mio;
 
-use std::io::{self, Read, Write};
-use std::sync::Arc;
+use std::io::{Read, Write};
+use std::net::TcpStream;
 use std::thread;
 
 use futures::Future;
 use futures::stream::Stream;
-use futures_io::copy;
-use futures_mio::TcpStream;
+use futures_io::{copy, TaskIo};
 
 macro_rules! t {
     ($e:expr) => (match $e {
@@ -30,8 +29,6 @@ fn echo_server() {
 
     let msg = "foo bar baz";
     let t = thread::spawn(move || {
-        use std::net::TcpStream;
-
         let mut s = TcpStream::connect(&addr).unwrap();
 
         for _i in 0..1024 {
@@ -44,10 +41,7 @@ fn echo_server() {
 
     let clients = srv.incoming();
     let client = clients.into_future().map(|e| e.0.unwrap()).map_err(|e| e.0);
-    let halves = client.map(|s| {
-        let s = Arc::new(s.0);
-        (SocketIo(s.clone()), SocketIo(s))
-    });
+    let halves = client.map(|s| TaskIo::new(s.0).split());
     let copied = halves.and_then(|(a, b)| copy(a, b));
 
     let amt = t!(l.run(copied));
@@ -55,21 +49,3 @@ fn echo_server() {
 
     assert_eq!(amt, msg.len() as u64 * 1024);
 }
-
-struct SocketIo(Arc<TcpStream>);
-
-impl Read for SocketIo {
-    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        (&*self.0).read(buf)
-    }
-}
-
-impl Write for SocketIo {
-    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        (&*self.0).write(buf)
-    }
-
-    fn flush(&mut self) -> io::Result<()> {
-        (&*self.0).flush()
-    }
-}

tests/stream-buffered.rs

@@ -3,14 +3,13 @@ extern crate futures_io;
 extern crate futures_mio;
 extern crate env_logger;
 
-use std::sync::Arc;
+use std::io::{Read, Write};
+use std::net::TcpStream;
 use std::thread;
-use std::io::{self, Read, Write};
 
 use futures::Future;
 use futures::stream::Stream;
-use futures_io::copy;
-use futures_mio::TcpStream;
+use futures_io::{copy, TaskIo};
 
 macro_rules! t {
     ($e:expr) => (match $e {
@@ -29,8 +28,6 @@ fn echo_server() {
     let addr = t!(srv.local_addr());
 
     let t = thread::spawn(move || {
-        use std::net::TcpStream;
-
         let mut s1 = t!(TcpStream::connect(&addr));
         let mut s2 = t!(TcpStream::connect(&addr));
 
@@ -45,8 +42,7 @@ fn echo_server() {
     });
 
     let future = srv.incoming()
-                    .map(|s| Arc::new(s.0))
-                    .map(|i| (SocketIo(i.clone()), SocketIo(i)))
+                    .map(|s| TaskIo::new(s.0).split())
                     .map(|(a, b)| copy(a, b).map(|_| ()))
                     .buffered(10)
                     .take(2)
@@ -56,21 +52,3 @@ fn echo_server() {
 
     t.join().unwrap();
 }
-
-struct SocketIo(Arc<TcpStream>);
-
-impl Read for SocketIo {
-    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        (&*self.0).read(buf)
-    }
-}
-
-impl Write for SocketIo {
-    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        (&*self.0).write(buf)
-    }
-
-    fn flush(&mut self) -> io::Result<()> {
-        (&*self.0).flush()
-    }
-}