HttpsConnector break body stream

client.rs

extern crate futures;
extern crate hyper;
extern crate hyper_tls;
extern crate tokio_core;

use futures::{Async, Future, Poll, Sink, Stream};
use std::error::Error;
use futures::future::ok;
use tokio_core::reactor::Core;
use hyper::Client;
use hyper::client::HttpConnector;
use hyper_tls::HttpsConnector;
use std::str::FromStr;

fn main() {
    let mut core = Core::new().unwrap();
    let connector = HttpsConnector::new(4, &core.handle()).unwrap();
    //let connector = HttpConnector::new(4, &core.handle());
    let client = Client::configure()
        .connector(connector)
        .build(&core.handle());

    let uri = hyper::Uri::from_str("http://127.0.0.1:9080/").unwrap();
    let (tx, body) = hyper::Body::pair();

    let p = Payload { i: 0 };

    let mut req: hyper::Request = hyper::Request::new(hyper::Post, uri);
    req.set_body(body);

    let req = client.request(req).and_then(|res| {
        println!("Response: {}", res.status());
        ok(())
    });

    core.run(req.join(tx.send_all(p.map_err(|_| unreachable!())).then(|_| ok(()))))
        .unwrap();
}

struct Payload {
    i: u64,
}

impl Stream for Payload {
    type Item = Result<hyper::Chunk, hyper::error::Error>;
    type Error = Box<Error>;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        if self.i < 5 {
            println!("POLL {}", self.i);
            self.i = self.i + 1;
            Ok(Async::Ready(Some(Ok(hyper::Chunk::from("hello")))))
        } else {
            println!("DONE");
            Ok(Async::Ready(None))
        }
    }
}