stream.rs

struct RecordIoConnection {
  buf: BytesMut,
  body: hyper::Body,
}


impl RecordIoConnection {
  pub fn connect(url: &str, buf_sz: usize) -> FutureResponse<Item=RecordIoConnection, Error=hyper::error::Error> {
    let response = create_hyper_get(); 
    response.map(|r| RecordIoConnection{ buf: BytesMut::with_capacity(buf_sz), body: r.body() } ))
  }
}

impl Stream for RecordIoConnection {
  Item=RecordIoMessage
  Error=RecordIoError
  fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
    if let Some(chunk) = try_ready!(self.body.poll()) {
      self.buf.put(chunk.get(..));
      if let Some(message) = try_parse_message(self.buf) {
        Ok(Async::Ready(message))
      } 
    }
    Ok(Async::NotReady)  
  }
}
}