buf: stream and iter helpers (#1011)

tokio-buf/tests/chain.rs

@@ -4,7 +4,6 @@ extern crate bytes;
 extern crate futures;
 extern crate tokio_buf;
 
-use bytes::Buf;
 use futures::Async::*;
 use tokio_buf::{BufStream, BufStreamExt};
 

tokio-buf/tests/iter.rs

@@ -0,0 +1,33 @@
+extern crate bytes;
+extern crate futures;
+extern crate tokio_buf;
+
+use futures::Async::*;
+use std::io::Cursor;
+use tokio_buf::{util, BufStream};
+
+#[macro_use]
+mod support;
+
+type Buf = Cursor<&'static [u8]>;
+
+#[test]
+fn empty_iter() {
+    let mut bs = util::iter(Vec::<Buf>::new());
+    assert_none!(bs.poll_buf());
+}
+
+#[test]
+fn full_iter() {
+    let bufs = vec![buf(b"one"), buf(b"two"), buf(b"three")];
+
+    let mut bs = util::iter(bufs);
+    assert_buf_eq!(bs.poll_buf(), "one");
+    assert_buf_eq!(bs.poll_buf(), "two");
+    assert_buf_eq!(bs.poll_buf(), "three");
+    assert_none!(bs.poll_buf());
+}
+
+fn buf(data: &'static [u8]) -> Buf {
+    Cursor::new(data)
+}

tokio-buf/tests/limit.rs

@@ -4,7 +4,6 @@ extern crate bytes;
 extern crate futures;
 extern crate tokio_buf;
 
-use bytes::Buf;
 use futures::Async::*;
 use futures::Future;
 use tokio_buf::{BufStream, BufStreamExt};

tokio-buf/tests/stream.rs

@@ -0,0 +1,49 @@
+extern crate bytes;
+extern crate futures;
+extern crate tokio_buf;
+extern crate tokio_mock_task;
+
+use futures::sync::mpsc;
+use futures::Async::*;
+use std::io::Cursor;
+use tokio_buf::{util, BufStream};
+use tokio_mock_task::MockTask;
+
+#[macro_use]
+mod support;
+
+type Buf = Cursor<&'static [u8]>;
+
+#[test]
+fn empty_stream() {
+    let (_, rx) = mpsc::unbounded::<Buf>();
+    let mut bs = util::stream(rx);
+    assert_none!(bs.poll_buf());
+}
+
+#[test]
+fn full_stream() {
+    let (tx, rx) = mpsc::unbounded();
+    let mut bs = util::stream(rx);
+    let mut task = MockTask::new();
+
+    tx.unbounded_send(buf(b"one")).unwrap();
+
+    assert_buf_eq!(bs.poll_buf(), "one");
+    task.enter(|| assert_not_ready!(bs.poll_buf()));
+
+    tx.unbounded_send(buf(b"two")).unwrap();
+
+    assert!(task.is_notified());
+    assert_buf_eq!(bs.poll_buf(), "two");
+    task.enter(|| assert_not_ready!(bs.poll_buf()));
+
+    drop(tx);
+
+    assert!(task.is_notified());
+    assert_none!(bs.poll_buf());
+}
+
+fn buf(data: &'static [u8]) -> Buf {
+    Cursor::new(data)
+}

tokio-buf/tests/string.rs

@@ -2,7 +2,6 @@ extern crate bytes;
 extern crate futures;
 extern crate tokio_buf;
 
-use bytes::Buf;
 use futures::Async::*;
 use std::fmt;
 use tokio_buf::BufStream;

tokio-buf/tests/support.rs

@@ -14,6 +14,7 @@ use std::io::Cursor;
 
 macro_rules! assert_buf_eq {
     ($actual:expr, $expect:expr) => {{
+        use bytes::Buf;
         match $actual {
             Ok(Ready(Some(val))) => {
                 assert_eq!(val.remaining(), val.bytes().len());