A lower level Connection API for the Client

[seanmonstar]

The Client is a higher level API that, among other things, provides a connection pool. Similar to how the server has higher and lower level APIs (Server vs Serve/Connection), the client should gain a lower level connection based API. This would allow finer grained control over when connections are made, and would allow external pool implementations (cc #1253).

Implementation

There would be a new hyper::client::Connection type, which implements Service. It would represent a single connection bound to HTTP. It would not reconnect itself when it is closed, someone managing the connection would handle that.

[sfackler]

This would be great!

[seanmonstar]

The API I've been prototyping has a similar look to the h2 client, where you provide an already connected io, and get back a sender (SendRequest?) and Connection pair. You probably want to just spawn the Connection into an executor, and then use the sender to send requests, but there are methods on Connection to allow finer control.

At this level, the internals can be configured to allow HTTP upgrades and CONNECTs (where as the higher level Client and Server treat them as errors, since there's no exposed API to handle them yet). After a user sends a request wanting an upgrade, and gets a 101 response back, they could coordinate to get the original io back out of the Connection. There's a few parts of this that may be tricky. Details next.

SendRequest

• This has a poll_ready method, allowing you to know if the related Connection is still usable.
• There is a send_request(&mut self, req: Request<B>) method, similar to Client::request(req).

Questions:

• What should this be called? SendRequest? Sender?
• What should the send_request method be called? send_request? request? send?
• Should this type be Clone?

Connection

• The easiest thing is to just have Connection::into_inner(self) -> Io, however, servers can send the 101 upgrade response and the first part of the new protocol in the same packet. If they do that, the internal state will likely have those bytes in its read buffer. So, just calling conn.into_inner() would mean losing data. For that reason, I'm wondering if it is better to just remove this possible footgun.

• The next thing that could be done is have Connection::into_parts(self) -> Parts<Io>. This is kind of like http's into_parts, which returns a struct with public fields, and a private one, allowing the possibility of adding new fields, while also allowing a user to take ownership of multiple things at once.

  pub struct Parts<T> {
      pub io: T,
      pub read_buf: Bytes,
      _non_exhaustive: (),
  }

• By default, the impl Future for Connection assumes it should manage the socket until completion, and then call AsyncWrite::shutdown(). However, if you're wanting to do an upgrade, you don't want shutdown to be called, but you do want to know when the Connection is otherwise "done" with HTTP. So, there probably needs to be another poll_* method to do that.

Questions:

• Should there be an into_inner() that could possibly lose data?
• If you were to call into_inner or into_parts at some arbitrary time, even if there wasn't an upgrade, there could be unfinished state internally. Should it be an error to do that? Or should that extra state be available the Parts struct?
• What should Parts be called? Is Parts too vague (hyper::client::conn::Parts)? I didn't want to say specifically Upgrade, since you might use it without an upgrade...
• What buffer should be returned in Parts? It is internally a BytesMut, but we may not want to expose that directly. It's not hard to convert almost anything into a Bytes, so I started with that...
• Should Parts include both the read_buf and write_buf?
• What should the secondary poll_ method be called, that resolves when HTTP is finished but without calling shutdown? Connection::poll_upgrade()? What should it return? Should it return an Error if the connection is closed without a successful upgrade? Or should that be an Ok result, just with an additional enum?

Example

Here's an example that I'm currently using in tests (with await inserted for terseness):

use hyper::client::conn;

// connect a tcpstream somehow, and then...
let (mut tx, mut conn) = await conn::handshake(io)?;
let until_upgrade = future::poll_fn(|| conn.poll_upgrade());

// using `http` crate
let req = Request::builder()
    .uri(uri)
    .header("upgrade", "foobar")
    .header("connection", "upgrade")
    .body(Body::empty())
    .unwrap();

let upgrade = tx.send_request(req)
    .map(|res| {
        assert_eq!(res.status(), StatusCode::SWITCHING_PROTOCOLS);
        assert_eq!(res.headers()["upgrade"], "foobar");
    });

await upgrade.join(until_upgrade)?;

let tcp = conn.into_parts().io;
// use tcp over 'foobar' protocol

[sfackler]

If we made Parts implement Read/Write that could be a decent way of handling upgrades in a non-footgunny way. You could manually grab the existing buffered data and stream out separately or just use the type itself as the stream for the upgraded protocol.

[seanmonstar]

If we made Parts implement Read/Write

That's a definite possibility!