Add read_some free function to read some bytes

src/io/mod.rs

@@ -35,6 +35,7 @@ mod copy;
 mod flush;
 mod read_exact;
 mod read_to_end;
+mod read;
 mod split;
 mod window;
 mod write_all;
@@ -42,6 +43,7 @@ pub use self::copy::{copy, Copy};
 pub use self::flush::{flush, Flush};
 pub use self::read_exact::{read_exact, ReadExact};
 pub use self::read_to_end::{read_to_end, ReadToEnd};
+pub use self::read::{read_some, ReadSome};
 pub use self::split::{ReadHalf, WriteHalf};
 pub use self::window::Window;
 pub use self::write_all::{write_all, WriteAll};

src/io/read.rs

@@ -0,0 +1,65 @@
+use std::mem;
+use std::io::Read;
+
+use futures::{Async, Future, Poll};
+
+enum State<R, T> {
+    Pending {
+        rd: R,
+        buf: T,
+    },
+    Empty,
+}
+
+fn eof() -> ::std::io::Error {
+    ::std::io::Error::new(::std::io::ErrorKind::UnexpectedEof, "unexpected EOF")
+}
+
+/// Baz.
+pub fn read_some<R, T>(rd: R, buf: T) -> ReadSome<R, T>
+    where R: Read,
+          T: AsMut<[u8]>
+{
+    ReadSome {
+        state: State::Pending {
+            rd: rd,
+            buf: buf,
+        }
+    }
+}
+
+/// Bar.
+pub struct ReadSome<R, T> {
+    state: State<R, T>,
+}
+
+impl<R, T> Future for ReadSome<R, T>
+    where R: Read,
+          T: AsMut<[u8]>
+{
+    type Item = (R, T, usize);
+    type Error = ::std::io::Error;
+
+    fn poll(&mut self) -> Poll<(R, T, usize), ::std::io::Error> {
+        let nread = match self.state {
+            State::Pending { ref mut rd, ref mut buf } => {
+                let buf = buf.as_mut();
+
+                match rd.read(&mut buf[..]) {
+                    Ok(0) => return Err(eof()),
+                    Ok(nread) => nread,
+                    Err(ref err) if err.kind() == ::std::io::ErrorKind::WouldBlock => {
+                        return Ok(Async::NotReady)
+                    }
+                    Err(err) => return Err(err.into()),
+                }
+            }
+            State::Empty => panic!("poll a ReadSome after it's done"),
+        };
+
+        match mem::replace(&mut self.state, State::Empty) {
+            State::Pending { rd, buf } => Ok((rd, buf, nread).into()),
+            State::Empty => panic!("invalid internal state"),
+        }
+    }
+}