codec: update to use std-future (#1214)

Strategy was to

- copy the old codec code that was temporarily being stashed in `tokio-io`
- modify all the type signatures to use Pin, as literal a translation as possible
- fix up the tests likewise

This is intended just to get things compiling and passing tests. Beyond that there is surely
lots of refactoring that can be done to make things more idiomatic. The docs are unchanged.

Closes #1189

tokio-codec/tests/codecs.rs

@@ -118,8 +118,8 @@ fn lines_decoder_max_length() {
 
     // Line that's one character too long. This could cause an out of bounds
     // error if we peek at the next characters using slice indexing.
-    // buf.put("aaabbbc");
-    // assert!(codec.decode(buf).is_err());
+    buf.put("aaabbbc");
+    assert!(codec.decode(buf).is_err());
 }
 
 #[test]

tokio-codec/tests/framed.rs

@@ -1,11 +1,16 @@
 #![deny(warnings, rust_2018_idioms)]
 
-use bytes::{Buf, BufMut, BytesMut, IntoBuf};
-use futures::{Future, Stream};
 use std::io::{self, Read};
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
 use tokio_codec::{Decoder, Encoder, Framed, FramedParts};
+use tokio_current_thread::block_on_all;
 use tokio_io::AsyncRead;
 
+use bytes::{Buf, BufMut, BytesMut, IntoBuf};
+use futures::prelude::{FutureExt, StreamExt};
+
 const INITIAL_CAPACITY: usize = 8 * 1024;
 
 /// Encode and decode u32 values.
@@ -49,7 +54,15 @@ impl Read for DontReadIntoThis {
     }
 }
 
-impl AsyncRead for DontReadIntoThis {}
+impl AsyncRead for DontReadIntoThis {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        _buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        unreachable!()
+    }
+}
 
 #[test]
 fn can_read_from_existing_buf() {
@@ -58,12 +71,12 @@ fn can_read_from_existing_buf() {
 
     let framed = Framed::from_parts(parts);
 
-    let num = framed
-        .into_future()
-        .map(|(first_num, _)| first_num.unwrap())
-        .wait()
-        .map_err(|e| e.0)
-        .unwrap();
+    let num = block_on_all(
+        framed
+            .into_future()
+            .map(|(first_num, _)| first_num.unwrap()),
+    )
+    .unwrap();
 
     assert_eq!(num, 42);
 }

tokio-codec/tests/framed_read.rs

@@ -1,12 +1,18 @@
 #![deny(warnings, rust_2018_idioms)]
 
-use bytes::{Buf, BytesMut, IntoBuf};
-use futures::Async::{NotReady, Ready};
-use futures::Stream;
 use std::collections::VecDeque;
 use std::io::{self, Read};
+use std::pin::Pin;
+use std::task::Poll::{Pending, Ready};
+use std::task::{Context, Poll};
+
+use bytes::{Buf, BytesMut, IntoBuf};
+use futures::Stream;
+
 use tokio_codec::{Decoder, FramedRead};
 use tokio_io::AsyncRead;
+use tokio_test::assert_ready;
+use tokio_test::task::MockTask;
 
 macro_rules! mock {
     ($($x:expr,)*) => {{
@@ -16,6 +22,19 @@ macro_rules! mock {
     }};
 }
 
+macro_rules! assert_read {
+    ($e:expr, $n:expr) => {{
+        let val = assert_ready!($e);
+        assert_eq!(val.unwrap().unwrap(), $n);
+    }};
+}
+
+macro_rules! pin {
+    ($id:ident) => {
+        Pin::new(&mut $id)
+    };
+}
+
 struct U32Decoder;
 
 impl Decoder for U32Decoder {
@@ -34,104 +53,146 @@ impl Decoder for U32Decoder {
 
 #[test]
 fn read_multi_frame_in_packet() {
+    let mut task = MockTask::new();
     let mock = mock! {
         Ok(b"\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x02".to_vec()),
     };
-
     let mut framed = FramedRead::new(mock, U32Decoder);
-    assert_eq!(Ready(Some(0)), framed.poll().unwrap());
-    assert_eq!(Ready(Some(1)), framed.poll().unwrap());
-    assert_eq!(Ready(Some(2)), framed.poll().unwrap());
-    assert_eq!(Ready(None), framed.poll().unwrap());
+
+    task.enter(|cx| {
+        assert_read!(pin!(framed).poll_next(cx), 0);
+        assert_read!(pin!(framed).poll_next(cx), 1);
+        assert_read!(pin!(framed).poll_next(cx), 2);
+        assert!(assert_ready!(pin!(framed).poll_next(cx)).is_none());
+    });
 }
 
 #[test]
 fn read_multi_frame_across_packets() {
+    let mut task = MockTask::new();
     let mock = mock! {
         Ok(b"\x00\x00\x00\x00".to_vec()),
         Ok(b"\x00\x00\x00\x01".to_vec()),
         Ok(b"\x00\x00\x00\x02".to_vec()),
     };
-
     let mut framed = FramedRead::new(mock, U32Decoder);
-    assert_eq!(Ready(Some(0)), framed.poll().unwrap());
-    assert_eq!(Ready(Some(1)), framed.poll().unwrap());
-    assert_eq!(Ready(Some(2)), framed.poll().unwrap());
-    assert_eq!(Ready(None), framed.poll().unwrap());
+
+    task.enter(|cx| {
+        assert_read!(pin!(framed).poll_next(cx), 0);
+        assert_read!(pin!(framed).poll_next(cx), 1);
+        assert_read!(pin!(framed).poll_next(cx), 2);
+        assert!(assert_ready!(pin!(framed).poll_next(cx)).is_none());
+    });
 }
 
 #[test]
 fn read_not_ready() {
+    let mut task = MockTask::new();
     let mock = mock! {
         Err(io::Error::new(io::ErrorKind::WouldBlock, "")),
         Ok(b"\x00\x00\x00\x00".to_vec()),
         Ok(b"\x00\x00\x00\x01".to_vec()),
     };
-
     let mut framed = FramedRead::new(mock, U32Decoder);
-    assert_eq!(NotReady, framed.poll().unwrap());
-    assert_eq!(Ready(Some(0)), framed.poll().unwrap());
-    assert_eq!(Ready(Some(1)), framed.poll().unwrap());
-    assert_eq!(Ready(None), framed.poll().unwrap());
+
+    task.enter(|cx| {
+        assert!(pin!(framed).poll_next(cx).is_pending());
+        assert_read!(pin!(framed).poll_next(cx), 0);
+        assert_read!(pin!(framed).poll_next(cx), 1);
+        assert!(assert_ready!(pin!(framed).poll_next(cx)).is_none());
+    });
 }
 
 #[test]
 fn read_partial_then_not_ready() {
+    let mut task = MockTask::new();
     let mock = mock! {
         Ok(b"\x00\x00".to_vec()),
         Err(io::Error::new(io::ErrorKind::WouldBlock, "")),
         Ok(b"\x00\x00\x00\x00\x00\x01\x00\x00\x00\x02".to_vec()),
     };
-
     let mut framed = FramedRead::new(mock, U32Decoder);
-    assert_eq!(NotReady, framed.poll().unwrap());
-    assert_eq!(Ready(Some(0)), framed.poll().unwrap());
-    assert_eq!(Ready(Some(1)), framed.poll().unwrap());
-    assert_eq!(Ready(Some(2)), framed.poll().unwrap());
-    assert_eq!(Ready(None), framed.poll().unwrap());
+
+    task.enter(|cx| {
+        assert!(pin!(framed).poll_next(cx).is_pending());
+        assert_read!(pin!(framed).poll_next(cx), 0);
+        assert_read!(pin!(framed).poll_next(cx), 1);
+        assert_read!(pin!(framed).poll_next(cx), 2);
+        assert!(assert_ready!(pin!(framed).poll_next(cx)).is_none());
+    });
 }
 
 #[test]
 fn read_err() {
+    let mut task = MockTask::new();
     let mock = mock! {
         Err(io::Error::new(io::ErrorKind::Other, "")),
     };
-
     let mut framed = FramedRead::new(mock, U32Decoder);
-    assert_eq!(io::ErrorKind::Other, framed.poll().unwrap_err().kind());
+
+    task.enter(|cx| {
+        assert_eq!(
+            io::ErrorKind::Other,
+            assert_ready!(pin!(framed).poll_next(cx))
+                .unwrap()
+                .unwrap_err()
+                .kind()
+        )
+    });
 }
 
 #[test]
 fn read_partial_then_err() {
+    let mut task = MockTask::new();
     let mock = mock! {
         Ok(b"\x00\x00".to_vec()),
         Err(io::Error::new(io::ErrorKind::Other, "")),
     };
-
     let mut framed = FramedRead::new(mock, U32Decoder);
-    assert_eq!(io::ErrorKind::Other, framed.poll().unwrap_err().kind());
+
+    task.enter(|cx| {
+        assert_eq!(
+            io::ErrorKind::Other,
+            assert_ready!(pin!(framed).poll_next(cx))
+                .unwrap()
+                .unwrap_err()
+                .kind()
+        )
+    });
 }
 
 #[test]
 fn read_partial_would_block_then_err() {
+    let mut task = MockTask::new();
     let mock = mock! {
         Ok(b"\x00\x00".to_vec()),
         Err(io::Error::new(io::ErrorKind::WouldBlock, "")),
         Err(io::Error::new(io::ErrorKind::Other, "")),
     };
-
     let mut framed = FramedRead::new(mock, U32Decoder);
-    assert_eq!(NotReady, framed.poll().unwrap());
-    assert_eq!(io::ErrorKind::Other, framed.poll().unwrap_err().kind());
+
+    task.enter(|cx| {
+        assert!(pin!(framed).poll_next(cx).is_pending());
+        assert_eq!(
+            io::ErrorKind::Other,
+            assert_ready!(pin!(framed).poll_next(cx))
+                .unwrap()
+                .unwrap_err()
+                .kind()
+        )
+    });
 }
 
 #[test]
 fn huge_size() {
+    let mut task = MockTask::new();
     let data = [0; 32 * 1024];
+    let mut framed = FramedRead::new(Slice(&data[..]), BigDecoder);
 
-    let mut framed = FramedRead::new(&data[..], BigDecoder);
-    assert_eq!(Ready(Some(0)), framed.poll().unwrap());
-    assert_eq!(Ready(None), framed.poll().unwrap());
+    task.enter(|cx| {
+        assert_read!(pin!(framed).poll_next(cx), 0);
+        assert!(assert_ready!(pin!(framed).poll_next(cx)).is_none());
+    });
 
     struct BigDecoder;
 
@@ -151,15 +212,19 @@ fn huge_size() {
 
 #[test]
 fn data_remaining_is_error() {
-    let data = [0; 5];
+    let mut task = MockTask::new();
+    let slice = Slice(&[0; 5]);
+    let mut framed = FramedRead::new(slice, U32Decoder);
 
-    let mut framed = FramedRead::new(&data[..], U32Decoder);
-    assert_eq!(Ready(Some(0)), framed.poll().unwrap());
-    assert!(framed.poll().is_err());
+    task.enter(|cx| {
+        assert_read!(pin!(framed).poll_next(cx), 0);
+        assert!(assert_ready!(pin!(framed).poll_next(cx)).unwrap().is_err());
+    });
 }
 
 #[test]
 fn multi_frames_on_eof() {
+    let mut task = MockTask::new();
     struct MyDecoder(Vec<u32>);
 
     impl Decoder for MyDecoder {
@@ -180,11 +245,14 @@ fn multi_frames_on_eof() {
     }
 
     let mut framed = FramedRead::new(mock!(), MyDecoder(vec![0, 1, 2, 3]));
-    assert_eq!(Ready(Some(0)), framed.poll().unwrap());
-    assert_eq!(Ready(Some(1)), framed.poll().unwrap());
-    assert_eq!(Ready(Some(2)), framed.poll().unwrap());
-    assert_eq!(Ready(Some(3)), framed.poll().unwrap());
-    assert_eq!(Ready(None), framed.poll().unwrap());
+
+    task.enter(|cx| {
+        assert_read!(pin!(framed).poll_next(cx), 0);
+        assert_read!(pin!(framed).poll_next(cx), 1);
+        assert_read!(pin!(framed).poll_next(cx), 2);
+        assert_read!(pin!(framed).poll_next(cx), 3);
+        assert!(assert_ready!(pin!(framed).poll_next(cx)).is_none());
+    });
 }
 
 // ===== Mock ======
@@ -207,4 +275,28 @@ impl Read for Mock {
     }
 }
 
-impl AsyncRead for Mock {}
+impl AsyncRead for Mock {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        match Pin::get_mut(self).read(buf) {
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => Pending,
+            other => Ready(other),
+        }
+    }
+}
+
+// TODO this newtype is necessary because `&[u8]` does not currently implement `AsyncRead`
+struct Slice<'a>(&'a [u8]);
+
+impl<'a> AsyncRead for Slice<'a> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        Ready(Pin::get_mut(self).0.read(buf))
+    }
+}

tokio-codec/tests/framed_write.rs

@@ -1,11 +1,17 @@
 #![deny(warnings, rust_2018_idioms)]
 
 use bytes::{BufMut, BytesMut};
-use futures::{Poll, Sink};
 use std::collections::VecDeque;
-use std::io::{self, Write};
 use tokio_codec::{Encoder, FramedWrite};
+use tokio_futures::Sink;
 use tokio_io::AsyncWrite;
+use tokio_test::assert_ready;
+use tokio_test::task::MockTask;
+
+use std::io::{self, Write};
+use std::pin::Pin;
+use std::task::Poll::{Pending, Ready};
+use std::task::{Context, Poll};
 
 macro_rules! mock {
     ($($x:expr,)*) => {{
@@ -15,6 +21,12 @@ macro_rules! mock {
     }};
 }
 
+macro_rules! pin {
+    ($id:ident) => {
+        Pin::new(&mut $id)
+    };
+}
+
 struct U32Encoder;
 
 impl Encoder for U32Encoder {
@@ -31,22 +43,28 @@ impl Encoder for U32Encoder {
 
 #[test]
 fn write_multi_frame_in_packet() {
+    let mut task = MockTask::new();
     let mock = mock! {
         Ok(b"\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x02".to_vec()),
     };
-
     let mut framed = FramedWrite::new(mock, U32Encoder);
-    assert!(framed.start_send(0).unwrap().is_ready());
-    assert!(framed.start_send(1).unwrap().is_ready());
-    assert!(framed.start_send(2).unwrap().is_ready());
 
-    // Nothing written yet
-    assert_eq!(1, framed.get_ref().calls.len());
+    task.enter(|cx| {
+        assert!(assert_ready!(pin!(framed).poll_ready(cx)).is_ok());
+        assert!(pin!(framed).start_send(0).is_ok());
+        assert!(assert_ready!(pin!(framed).poll_ready(cx)).is_ok());
+        assert!(pin!(framed).start_send(1).is_ok());
+        assert!(assert_ready!(pin!(framed).poll_ready(cx)).is_ok());
+        assert!(pin!(framed).start_send(2).is_ok());
 
-    // Flush the writes
-    assert!(framed.poll_complete().unwrap().is_ready());
+        // Nothing written yet
+        assert_eq!(1, framed.get_ref().calls.len());
 
-    assert_eq!(0, framed.get_ref().calls.len());
+        // Flush the writes
+        assert!(assert_ready!(pin!(framed).poll_flush(cx)).is_ok());
+
+        assert_eq!(0, framed.get_ref().calls.len());
+    });
 }
 
 #[test]
@@ -59,7 +77,7 @@ fn write_hits_backpressure() {
         Ok(b"".to_vec()),
     };
 
-    for i in 0..(ITER + 1) {
+    for i in 0..=ITER {
         let mut b = BytesMut::with_capacity(4);
         b.put_u32_be(i as u32);
 
@@ -70,7 +88,7 @@ fn write_hits_backpressure() {
                 if data.len() < ITER {
                     data.extend_from_slice(&b[..]);
                     continue;
-                }
+                } // else fall through and create a new buffer
             }
             _ => unreachable!(),
         }
@@ -78,27 +96,38 @@ fn write_hits_backpressure() {
         // Push a new new chunk
         mock.calls.push_back(Ok(b[..].to_vec()));
     }
+    // 1 'wouldblock', 4 * 2KB buffers, 1 b-byte buffer
+    assert_eq!(mock.calls.len(), 6);
 
+    let mut task = MockTask::new();
     let mut framed = FramedWrite::new(mock, U32Encoder);
+    task.enter(|cx| {
+        // Send 8KB. This fills up FramedWrite2 buffer
+        for i in 0..ITER {
+            assert!(assert_ready!(pin!(framed).poll_ready(cx)).is_ok());
+            assert!(pin!(framed).start_send(i as u32).is_ok());
+        }
 
-    for i in 0..ITER {
-        assert!(framed.start_send(i as u32).unwrap().is_ready());
-    }
+        // Now we poll_ready which forces a flush. The mock pops the front message
+        // and decides to block.
+        assert!(pin!(framed).poll_ready(cx).is_pending());
 
-    // This should reject
-    assert!(!framed.start_send(ITER as u32).unwrap().is_ready());
+        // We poll again, forcing another flush, which this time succeeds
+        // The whole 8KB buffer is flushed
+        assert!(assert_ready!(pin!(framed).poll_ready(cx)).is_ok());
 
-    // This should succeed and start flushing the buffer.
-    assert!(framed.start_send(ITER as u32).unwrap().is_ready());
+        // Send more data. This matches the final message expected by the mock
+        assert!(pin!(framed).start_send(ITER as u32).is_ok());
 
-    // Flush the rest of the buffer
-    assert!(framed.poll_complete().unwrap().is_ready());
+        // Flush the rest of the buffer
+        assert!(assert_ready!(pin!(framed).poll_flush(cx)).is_ok());
 
-    // Ensure the mock is empty
-    assert_eq!(0, framed.get_ref().calls.len());
+        // Ensure the mock is empty
+        assert_eq!(0, framed.get_ref().calls.len());
+    })
 }
 
-// ===== Mock ======
+// // ===== Mock ======
 
 struct Mock {
     calls: VecDeque<io::Result<Vec<u8>>>,
@@ -123,7 +152,23 @@ impl Write for Mock {
 }
 
 impl AsyncWrite for Mock {
-    fn shutdown(&mut self) -> Poll<(), io::Error> {
-        Ok(().into())
+    fn poll_write(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, io::Error>> {
+        match Pin::get_mut(self).write(buf) {
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => Pending,
+            other => Ready(other),
+        }
+    }
+    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        match Pin::get_mut(self).flush() {
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => Pending,
+            other => Ready(other),
+        }
+    }
+    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        unimplemented!()
     }
 }