Add file test case for streaming. Fix streaming responder.

resolves #3

examples/stream/src/main.rs

@@ -6,7 +6,8 @@ extern crate rocket;
 use rocket::Rocket;
 use rocket::response::{data, Stream};
 
-use std::io::{repeat, Repeat, Read, Take};
+use std::io::{self, repeat, Repeat, Read, Take};
+use std::fs::File;
 
 type LimitedRepeat = Take<Repeat>;
 
@@ -15,6 +16,13 @@ fn root() -> data::Plain<Stream<LimitedRepeat>> {
     data::Plain(Stream::from(repeat('a' as u8).take(25000)))
 }
 
-fn main() {
-    Rocket::new("localhost", 8000).mount_and_launch("/", routes![root]);
+#[get("/big_file")]
+fn file() -> io::Result<Stream<File>> {
+    // Generate this file using: head -c BYTES /dev/random > big_file.dat
+    const FILENAME: &'static str = "big_file.dat";
+    File::open(FILENAME).map(|file| Stream::from(file))
+}
+
+fn main() {
+    Rocket::new("localhost", 8000).mount_and_launch("/", routes![root, file]);
 }

lib/src/response/named_file.rs

@@ -14,6 +14,17 @@ impl NamedFile {
         Ok(NamedFile(path.as_ref().to_path_buf(), file))
     }
 
+    #[inline(always)]
+    pub fn file(&self) -> &File {
+        &self.1
+    }
+
+    #[inline(always)]
+    pub fn file_mut(&mut self) -> &mut File {
+        &mut self.1
+    }
+
+    #[inline(always)]
     pub fn path(&self) -> &Path {
         self.0.as_path()
     }
@@ -29,7 +40,7 @@ impl Responder for NamedFile {
             }
         }
 
-        self.1.respond(res)
+        self.file_mut().respond(res)
     }
 }
 
@@ -47,3 +58,50 @@ impl DerefMut for NamedFile {
     }
 }
 
+impl io::Read for NamedFile {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        self.file().read(buf)
+    }
+
+    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        self.file().read_to_end(buf)
+    }
+}
+
+impl io::Write for NamedFile {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.file().write(buf)
+    }
+
+    fn flush(&mut self) -> io::Result<()> { self.file().flush() }
+}
+
+impl io::Seek for NamedFile {
+    fn seek(&mut self, pos: io::SeekFrom) -> io::Result<u64> {
+        self.file().seek(pos)
+    }
+}
+
+impl<'a> io::Read for &'a NamedFile {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        self.file().read(buf)
+    }
+
+    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
+        self.file().read_to_end(buf)
+    }
+}
+
+impl<'a> io::Write for &'a NamedFile {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.file().write(buf)
+    }
+
+    fn flush(&mut self) -> io::Result<()> { self.file().flush() }
+}
+
+impl<'a> io::Seek for &'a NamedFile {
+    fn seek(&mut self, pos: io::SeekFrom) -> io::Result<u64> {
+        self.file().seek(pos)
+    }
+}

lib/src/response/stream.rs

@@ -1,15 +1,26 @@
 use response::*;
 use std::io::{Read, Write, ErrorKind};
 
-/// The size of each chunk in the streamed response.
+// TODO: Support custom chunk sizes.
+/// The default size of each chunk in the streamed response.
 pub const CHUNK_SIZE: usize = 4096;
 
-pub struct Stream<T: Read>(pub Box<T>);
+pub struct Stream<T: Read>(Box<T>);
 
 impl<T: Read> Stream<T> {
     pub fn from(reader: T) -> Stream<T> {
         Stream(Box::new(reader))
     }
+
+//     pub fn chunked(mut self, size: usize) -> Self {
+//         self.1 = size;
+//         self
+//     }
+
+//     #[inline(always)]
+//     pub fn chunk_size(&self) -> usize {
+//         self.1
+//     }
 }
 
 impl<T: Read> Responder for Stream<T> {
@@ -18,13 +29,11 @@ impl<T: Read> Responder for Stream<T> {
         let mut buffer = [0; CHUNK_SIZE];
         let mut complete = false;
         while !complete {
-            let mut left = CHUNK_SIZE;
-            while left > 0 && !complete {
-                match self.0.read(&mut buffer[..left]) {
+            let mut read = 0;
+            while read < buffer.len() && !complete {
+                match self.0.read(&mut buffer[read..]) {
                     Ok(n) if n == 0 => complete = true,
-                    Ok(n) if n < left => left -= n,
-                    Ok(n) if n == left => left = CHUNK_SIZE,
-                    Ok(n) => unreachable!("Impossible byte count {}/{}!", n, left),
+                    Ok(n) => read += n,
                     Err(ref e) if e.kind() == ErrorKind::Interrupted => continue,
                     Err(ref e) => {
                         error_!("Error streaming response: {:?}", e);
@@ -33,7 +42,7 @@ impl<T: Read> Responder for Stream<T> {
                 }
             }
 
-            if let Err(e) = stream.write_all(&buffer) {
+            if let Err(e) = stream.write_all(&buffer[..read]) {
                 error_!("Stream write_all() failed: {:?}", e);
                 return Outcome::FailStop;
             }