[WIP] Improve HTTP/1.1 upgrade support

[ubnt-intrepid]

let protocol = hyper::server::conn::Http::new();
...
let mut conn = Some(protocol.serve_connection(stream, service));

// impl Future<Item = Option<Parts<I, S>>, Error = hyper::Error>
let conn_without_shutdown = futures::future::poll_fn(move || -> hyper::Result<_> {
    futures::try_ready!(conn.as_mut().unwrap().poll_without_shutdown());
    Ok(Async::Ready(conn.take().unwrap().try_into_parts()))
})

conn_without_shutdown.then(|res| {
    match res {
        Ok(Some(Parts { io, read_buf, service, .. })) => {
            futures::future::Either::A(service.into_upgrade(io, read_buf))
        }
        Ok(None) => futures::future::Either::B(futures::future::ok(())),
        Err(e) => futures::future::Either::B(futures::future::err(e))),
    }
})

Pros:

• the future representing the post-upgrade process (i.e. HttpUpgrade::Upgraded) is driven on the task same as Connection belongs. It means that it is easier to control the behavior of task per connection (timeout, etc.)
  • when using hyper::upgrade, the post-upgrade process is located on a spawned task by the user and different from the task of Connection.

Cons:

• HttpService and HttpConnection will become complicated due to abstraction around upgrade mechanism. This might affect the design of Web frameworks.

[ubnt-intrepid]

Candidate 1

modification in HttpConnection:

trait HttpConnection<I: AsyncRead + AsyncWrite> {
    type Response: HttpResponse;
    type Upgrade: HttpUpgrade<I>;
    type Error;
    type Future: Future<Item = Either<Self::Response, Self::Upgrade>, Error = Self::Error>;

    fn respond(&mut self, request: HttpRequest) -> Self::Future;
}

where HttpUpgrade is defined as follows:

trait HttpUpgrade<I: AsyncRead + AsyncWrite> {
    type Task: HttpUpgradeTask<I>;

    fn into_response(self) -> (Response<()>, Self::Task);
}

trait HttpUpgradeTask<I> {
    type Future: Future<Item = (), Error = ()>;
    fn upgrade(self, io: Upgraded<I>) -> Self::Future;
}

struct Upgraded<I> {
    io: I,
    read_buf: Bytes,
}
// impl io::Read, io::Write, tokio::io::AsyncRead, tokio::io::AsyncWrite, ...

struct InnerService<S: HttpService<I>, I: AsyncRead + AsyncWrite> {
    connection: S::Connection,
    ...
    rx_upgraded: Option<oneshot::Receiver<<S::Connection as HttpConnection<I>>::Upgrade>>,
}

[ubnt-intrepid]

HttpUpgrade was added to izanami_http.