All client connections are leaked when keep_alive(false)

[sfackler]

extern crate hyper;
extern crate tokio_core;
extern crate tokio_io;
extern crate futures;
extern crate env_logger;

use tokio_core::reactor::{Core, Timeout};
use tokio_core::net::TcpStream;
use tokio_io::{AsyncRead, AsyncWrite};
use hyper::{Client, Uri};
use hyper::client::HttpConnector;
use hyper::server::Service;
use std::time::Duration;
use std::cell::Cell;
use futures::{Future, Stream, stream};
use std::io::{self, Read, Write};

fn main() {
    env_logger::init().unwrap();

    let mut core = Core::new().unwrap();
    let handle = core.handle();

    let client = Client::configure()
        .connector(DebugConnector(HttpConnector::new(1, &handle), Cell::new(0)))
        .no_proto()
        .keep_alive(false)
        .build(&handle);

    let fut = stream::repeat(()).for_each(|()| {
        client.get("http://www.google.com".parse().unwrap())
            .and_then(|resp| {
                resp.body().collect()
            })
            .and_then(|_| {
                let timeout = Timeout::new(Duration::from_secs(1), &handle).unwrap();
                timeout.map_err(|e| e.into())
            })
            .map_err(|e| println!("error {}", e))
    });

    core.run(fut).unwrap();
}

struct DebugConnector(HttpConnector, Cell<u32>);

impl Service for DebugConnector {
    type Request = Uri;
    type Response = DebugStream;
    type Error = io::Error;
    type Future = Box<Future<Item = DebugStream, Error = io::Error>>;

    fn call(&self, uri: Uri) -> Self::Future {
        let id = self.1.get();
        self.1.set(id + 1);
        println!("new connection {}", id);
        Box::new(self.0.call(uri).map(move |s| DebugStream(s, id)))
    }
}

struct DebugStream(TcpStream, u32);

impl Drop for DebugStream {
    fn drop(&mut self) {
        println!("socket {} dropping", self.1);
    }
}

impl Write for DebugStream {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.0.write(buf)
    }

    fn flush(&mut self) -> io::Result<()> {
        self.0.flush()
    }
}

impl AsyncWrite for DebugStream {
    fn shutdown(&mut self) -> futures::Poll<(), io::Error> {
        AsyncWrite::shutdown(&mut self.0)
    }
}

impl Read for DebugStream {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        self.0.read(buf)
    }
}

impl AsyncRead for DebugStream {}

~/foo ❯ cargo run                                                                                                                                                                            [foo/master +%]
   Compiling foo v0.1.0 (file:///Users/sfackler/foo)
    Finished dev [unoptimized + debuginfo] target(s) in 5.28 secs
     Running `target/debug/foo`
new connection 0
new connection 1
new connection 2
new connection 3
new connection 4
new connection 5
new connection 6
new connection 7
new connection 8
new connection 9
new connection 10
new connection 11
new connection 12
new connection 13
new connection 14
...

This happens both with and without no_proto