fix(conn): always flush io from poll_complete

Closes #1108

src/http/conn.rs

@@ -827,6 +827,7 @@ mod tests {
             assert!(conn.state.writing.is_queued());
             assert!(conn.poll_complete().unwrap().is_ready());
             assert!(!conn.state.writing.is_queued());
+            assert!(conn.io.io_mut().flushed());
 
             Ok(())
         }).wait();

src/http/io.rs

@@ -134,15 +134,16 @@ impl<T: Write> Write for Buffered<T> {
 
     fn flush(&mut self) -> io::Result<()> {
         if self.write_buf.remaining() == 0 {
-            Ok(())
+            self.io.flush()
         } else {
             loop {
                 let n = try!(self.write_buf.write_into(&mut self.io));
                 debug!("flushed {} bytes", n);
                 if self.write_buf.remaining() == 0 {
-                    return Ok(())
+                    break;
                 }
             }
+            self.io.flush()
         }
     }
 }

src/mock.rs

@@ -62,6 +62,7 @@ pub struct AsyncIo<T> {
     inner: T,
     bytes_until_block: usize,
     error: Option<io::Error>,
+    flushed: bool,
 }
 
 impl<T> AsyncIo<T> {
@@ -70,6 +71,7 @@ impl<T> AsyncIo<T> {
             inner: inner,
             bytes_until_block: bytes,
             error: None,
+            flushed: false,
         }
     }
 
@@ -86,6 +88,10 @@ impl AsyncIo<Buf> {
     pub fn new_buf<T: Into<Vec<u8>>>(buf: T, bytes: usize) -> AsyncIo<Buf> {
         AsyncIo::new(Buf::wrap(buf.into()), bytes)
     }
+
+    pub fn flushed(&self) -> bool {
+        self.flushed
+    }
 }
 
 impl<T: Read> Read for AsyncIo<T> {
@@ -111,6 +117,7 @@ impl<T: Write> Write for AsyncIo<T> {
             Err(io::Error::new(io::ErrorKind::WouldBlock, "mock block"))
         } else {
             trace!("AsyncIo::write() block_in = {}, data.len() = {}", self.bytes_until_block, data.len());
+            self.flushed = false;
             let n = cmp::min(self.bytes_until_block, data.len());
             let n = try!(self.inner.write(&data[..n]));
             self.bytes_until_block -= n;
@@ -119,6 +126,7 @@ impl<T: Write> Write for AsyncIo<T> {
     }
 
     fn flush(&mut self) -> io::Result<()> {
+        self.flushed = true;
         self.inner.flush()
     }
 }