futures: fix sink usage

when start_send returns NotReady you're supposed to call poll_complete before bubbling that up

Signed-off-by: Marc-Antoine Perennou <[email protected]>

futures/src/transport.rs

@@ -195,6 +195,7 @@ impl<T> AMQPTransport<T>
 
   /// Poll the network to send outcoming frames.
   fn poll_send(&mut self) -> Poll<(), io::Error> {
+    let mut ret = Ok(Async::Ready(()));
     while let Some(frame) = self.conn.next_frame() {
       trace!("transport poll_send; frame={:?}", frame);
       match self.start_send(frame)? {
@@ -204,11 +205,16 @@ impl<T> AMQPTransport<T>
         AsyncSink::NotReady(frame) => {
           trace!("transport poll_send; status=NotReady");
           self.conn.frame_queue.push_front(frame);
-          return Ok(Async::NotReady);
+          ret = Ok(Async::NotReady);
+          break;
         }
       }
     }
-    self.poll_complete()
+    if self.poll_complete()?.is_ready() {
+      ret
+    } else {
+      Ok(Async::NotReady)
+    }
   }
 }