///|
/// An HTTP transport with bounded IO operations and a small keep-alive pool.
///
/// A connection returns to the pool only after its response body has been
/// consumed in full and the server did not answer `Connection: close`. At most
/// `max_idle_per_origin` connections are parked per origin, each reusable for
/// at most `idle_max_ms`; concurrent requests beyond that dial new connections.
/// They queue only under `max_connections` or `max_connections_per_origin`,
/// which cap the connections in use (parked ones do not count). No timer
/// runs: every request, to any origin, closes the parked connections that
/// have expired, so a transport left unused holds them until its next request
/// or `close()`. When a reused connection fails
/// before a response head, the request is sent once more on a fresh connection
/// if it is idempotent or carries an `Idempotency-Key` and its body is not a
/// stream already read; other requests surface the failure.
///
/// A body in memory is sent with `Content-Length`; so is a stream body
/// declared with a length (`Request::body_stream`), while one without is sent
/// `Transfer-Encoding: chunked`. A body is written in slices of at most
/// 64 KiB, each as it is read from a stream, which is taken once and closed
/// when the request is sent or fails. On JS, moonbitlang/async hands fetch
/// the body whole, so a stream body is read into memory first and one over
/// 16 MiB fails with `Protocol` before anything is sent; send larger bodies
/// as bytes there.
///
/// On native and Wasm, a request without `Accept-Encoding` is sent with
/// `Accept-Encoding: gzip, deflate, br` and its response body is decoded: the
/// codings undone leave `Content-Encoding`, `Content-Length` is dropped, and an
/// unknown coding stays on the body untouched. A request that sets
/// `Accept-Encoding` itself gets the body as sent. On JS, fetch does both.
pub struct AsyncTransport {
priv limits : Timeouts
priv clock : &@clock.Clock
priv max_body_bytes : Int32?
priv pool : IdlePool[@ahttp.Client]
priv slots : Slots
priv proxies : Proxies
}
///|
/// Creates a transport.
///
/// Time limits, in milliseconds; a nonpositive one fails at once:
/// - `timeout_ms` bounds an exchange from the call, slot wait included: a
/// buffered one to the end of its body, a streaming one to its head. Every
/// phase ends at that deadline at the latest, reporting `ms=timeout_ms`.
/// - `connect_timeout_ms` bounds dialing a new connection: DNS, TCP and TLS.
/// - `write_timeout_ms` bounds sending the request head, and each write of
/// at most 64 KiB of the body on its own; for a stream body, each read from
/// the stream as well.
/// - `read_timeout_ms` bounds waiting for the head, and each body read on its
/// own. It alone bounds a stream's reads.
/// - `pool_timeout_ms` bounds the wait for a slot under a connection cap.
///
/// Each phase limit defaults to `timeout_ms`, which keeps one limit for the
/// exchange and one per stream read. A timeout names its phase: `Pool`,
/// `Dial`, `Write`, `Head` or `Body`. On JS, fetch connects, sends and waits
/// for the head in one call, under `read_timeout_ms` and as phase `Unknown`;
/// the connect and write limits do not apply there.
///
/// `max_connections` caps the connections in use across origins, and
/// `max_connections_per_origin` those to one origin: a connection is in use
/// from its request's slot until its body is read or abandoned. A request
/// over a cap waits for a slot, first come first served among the waiters a
/// freed slot fits, until `pool_timeout_ms`; caps below 1 count as 1. Unset,
/// requests never wait.
///
/// `max_idle_per_origin = 0` or `idle_max_ms = 0` disables connection reuse.
/// `max_body_bytes` bounds a buffered response body, counted after decoding:
/// a longer one closes the connection and fails with `Protocol`. Without it
/// the body is unbounded. Streams are never bounded, since their reader
/// decides what to keep.
///
/// `proxy`, an `http://` or `https://` URL with an optional
/// `user:password@`, carries every request: an `http://` one is forwarded
/// to it, an `https://` one goes through a `CONNECT` tunnel, and the
/// credentials go to the proxy alone, as `Basic` `Proxy-Authorization`.
/// Without it, and unless `trust_env` is false, the transport takes its
/// proxies from `HTTP_PROXY`, `HTTPS_PROXY`, `ALL_PROXY` and `NO_PROXY`
/// (lower case first) as curl and httpx read them, once, here. Each request
/// is routed by its own URL. A proxy it cannot reach fails the request with
/// `Connect`, a refused tunnel with `HttpError::Proxy`, and an unusable
/// proxy URL with `Protocol`; none of them sends it direct. On JS, fetch
/// uses the platform's proxy settings, if any, and the environment is not
/// read: a request on a transport given `proxy` fails with `Protocol`.
///
/// `clock` measures every limit above and the idle time of parked
/// connections; a test passes a virtual clock to expire them without
/// waiting.
///
/// ```mbt check
/// test {
/// let _ = @adapter.AsyncTransport::new(timeout_ms=1000)
/// let _ = @adapter.AsyncTransport::new(
/// connect_timeout_ms=5000,
/// read_timeout_ms=60000,
/// )
/// let _ = @adapter.AsyncTransport::new(max_connections_per_origin=8)
/// let _ = @adapter.AsyncTransport::new(max_idle_per_origin=0)
/// let _ = @adapter.AsyncTransport::new(max_body_bytes=1_048_576)
/// let _ = @adapter.AsyncTransport::new(proxy="http://user:pass@proxy:3128")
/// let _ = @adapter.AsyncTransport::new(trust_env=false)
/// }
/// ```
pub fn AsyncTransport::new(
timeout_ms~ : Int32 = 30000,
connect_timeout_ms? : Int32,
write_timeout_ms? : Int32,
read_timeout_ms? : Int32,
pool_timeout_ms? : Int32,
max_connections? : Int32,
max_connections_per_origin? : Int32,
max_idle_per_origin~ : Int32 = 4,
idle_max_ms~ : Int32 = 10000,
max_body_bytes? : Int32,
proxy? : String,
trust_env~ : Bool = true,
clock~ : &@clock.Clock = AsyncClock::new(),
) -> AsyncTransport {
{
limits: {
total: timeout_ms,
pool: pool_timeout_ms.unwrap_or(timeout_ms),
connect: connect_timeout_ms.unwrap_or(timeout_ms),
write: write_timeout_ms.unwrap_or(timeout_ms),
read: read_timeout_ms.unwrap_or(timeout_ms),
},
clock,
max_body_bytes,
pool: IdlePool::new(max_per_origin=max_idle_per_origin, idle_max_ms~),
slots: Slots::new(
max_total?=max_connections,
max_per_origin?=max_connections_per_origin,
),
proxies: platform_proxies(proxy, trust_env~, @env.get_env_var),
}
}
///|
/// Explicit transport methods for buffered and streaming requests.
pub extend AsyncTransport with @http.Transport::{send, send_stream}
///|
/// Reads the entire response within the exchange timeout, regardless of status.
pub impl @http.Transport for AsyncTransport with fn send(self, request) {
let timer = PhaseTimer::exchange(self.clock, self.limits)
let opened = self.open_request(request, timer)
errdefer self.discard(opened.key, opened.client)
let body = read_body(
opened.client,
opened.decoder,
self.max_body_bytes,
timer,
)
self.release(opened)
@http.Response::{
status: opened.head.status,
headers: opened.head.headers,
body,
}
}
///|
/// Reads the response head within the exchange timeout and transfers
/// ownership of the connection to a stream.
pub impl @http.Transport for AsyncTransport with fn send_stream(self, request) {
let exchange = self.open_request(
request,
PhaseTimer::exchange(self.clock, self.limits),
)
let stream : &@http.BodyStream = ClientStream::{
transport: self,
key: exchange.key,
user_agent: exchange.user_agent,
client: exchange.client,
decoder: exchange.decoder,
keep_alive: exchange.keep_alive,
timer: PhaseTimer::stream(self.clock, self.limits),
finished: false,
closed: false,
}
(exchange.head, stream)
}
///|
/// The most bytes one read takes from a connection. Without `max_len`, a read
/// returns whatever the connection has buffered, so a stream's chunk, and the
/// overshoot past `max_body_bytes` before it is refused, would be bounded only
/// by that buffer. A decoded read is cut to the same size.
let max_read_bytes : Int32 = 64 * 1024
///|
/// The next chunk of a body: the next `read`, or with a `decoder`, the next
/// decoded slice, reading only as much as that slice needs.
async fn read_decoded(
read : async () -> Bytes?,
decoder : ContentDecoder?,
) -> Bytes? {
guard decoder is Some(decoder) else { return read() }
for ;; {
match decoder.pull() {
Slice(slice) => return Some(slice)
Done => return None
NeedInput =>
match read() {
Some(chunk) => decoder.feed(chunk)
None => decoder.end()
}
}
}
}
///|
/// The next chunk of a response body as one `Body` phase of `timer`.
async fn read_chunk(
client : @ahttp.Client,
decoder : ContentDecoder?,
timer : PhaseTimer,
) -> Bytes? raise @http.HttpError {
timer.run(Body, () => {
read_decoded(() => client.read_some(max_len=max_read_bytes), decoder)
})
}
///|
/// The rest of a response body, decoded, refused once it exceeds `max_bytes`.
async fn read_body(
client : @ahttp.Client,
decoder : ContentDecoder?,
max_bytes : Int32?,
timer : PhaseTimer,
) -> Bytes raise @http.HttpError {
read_bounded(
() => read_chunk(client, decoder, timer),
max_bytes.unwrap_or(2147483647), // no limit short of `Int32`
) catch {
error => raise failure(Body, error)
}
}
///|
/// Chunks from `read` until it returns `None`. The chunk that would take the
/// total past `max_bytes` raises `Protocol` without reading further.
async fn read_bounded(read : async () -> Bytes?, max_bytes : Int32) -> Bytes {
let body = Buffer()
while read() is Some(chunk) {
if chunk.length() > max_bytes - body.length() {
raise @http.HttpError::Protocol(
"response body larger than \{max_bytes} bytes",
)
}
body.write_bytes(chunk)
}
body.to_bytes()
}
///|
priv struct ClientStream {
transport : AsyncTransport
key : String
user_agent : String?
client : @ahttp.Client
decoder : ContentDecoder?
keep_alive : Bool
timer : PhaseTimer
mut finished : Bool
mut closed : Bool
}
///|
/// Parks the connection when the body was read to its end; closes it when
/// the stream is abandoned early.
impl @http.BodyStream for ClientStream with fn close(self) {
if !self.closed {
self.closed = true
if self.finished && self.keep_alive {
self.transport.checkin(self.key, self.user_agent, self.client)
self.transport.slots.release(self.key)
} else {
self.transport.discard(self.key, self.client)
}
}
}
///|
impl @http.BodyStream for ClientStream with fn read_some(self) {
if self.closed {
return None
}
errdefer @http.BodyStream::close(self)
let chunk = read_chunk(self.client, self.decoder, self.timer)
if chunk is None {
self.finished = true
@http.BodyStream::close(self)
}
chunk
}
///|
/// One exchange whose response head has arrived; the body is still on
/// `client`, a connection parked and counted under `key` (`Route::key`).
priv struct Opened {
head : @http.ResponseHead
client : @ahttp.Client
decoder : ContentDecoder?
key : String
user_agent : String?
keep_alive : Bool
}
///|
/// Parks or closes the connection of a fully consumed exchange and frees its
/// slot.
fn AsyncTransport::release(self : AsyncTransport, opened : Opened) -> Unit {
if opened.keep_alive {
self.checkin(opened.key, opened.user_agent, opened.client)
self.slots.release(opened.key)
} else {
self.discard(opened.key, opened.client)
}
}
///|
/// Closes the connection of an exchange that ends early and frees its slot.
fn AsyncTransport::discard(
self : AsyncTransport,
key : String,
client : @ahttp.Client,
) -> Unit {
client.close()
self.slots.release(key)
}
///|
/// The request headers as sent, with `User-Agent` pulled out because
/// connections are dialed and reused per user agent. `Content-Length` and
/// `Transfer-Encoding` are dropped since the transport frames the body
/// itself, and `Proxy-Authorization` since proxy credentials come from the
/// proxy URL; repeated names are comma-joined.
fn wire_headers(headers : @http.Headers) -> (@ahttp.Headers, String?) {
let wire : @ahttp.Headers = Map([])
let mut user_agent : String? = None
for (name, value) in headers.iter() {
// The transport frames the body itself, and proxy credentials come from
// the proxy URL alone: a caller's `Proxy-Authorization` would otherwise
// reach the origin through a tunnel, or after a redirect.
if name == "content-length" ||
name == "transfer-encoding" ||
name == "proxy-authorization" {
continue
}
if name == "user-agent" {
user_agent = Some(
match user_agent {
None => value
Some(previous) => previous + ", " + value
},
)
continue
}
let key = @ahttp.CaseInsensitiveString(name)
wire[key] = match wire.get(key) {
None => value
Some(previous) => previous + ", " + value
}
}
(wire, user_agent)
}
///|
/// Takes a connection slot, sends `request` on the route its URL picks and
/// reads its response head, each phase under `timer`. The slot passes to the
/// returned exchange.
async fn AsyncTransport::open_request(
self : AsyncTransport,
request : @http.Request,
timer : PhaseTimer,
) -> Opened raise @http.HttpError {
let meth = request_method(request.http_method)
let (origin, path) = split_url(request.url)
let (headers, user_agent) = wire_headers(request.headers)
let negotiated = negotiates_content_coding &&
!request.headers.contains("accept-encoding")
if negotiated {
headers[@ahttp.CaseInsensitiveString("Accept-Encoding")] = accept_encoding()
}
if declares_content_length &&
content_length(request.http_method, request.body) is Some(length) {
headers[@ahttp.CaseInsensitiveString("Content-Length")] = length.to_string()
}
let route = self.proxies.route(origin)
refuse_unbufferable(request.body, buffered_stream_limit)
if self.pool.closed {
raise @http.HttpError::Connect("transport closed")
}
let key = route.key(origin.text)
let target = route.target(origin.text, path)
self.slots.acquire(key, timer)
errdefer self.slots.release(key)
let mut fresh = false
for ;; {
let (client, reused) = self.checkout(
key,
origin.text,
route,
user_agent,
fresh~,
timer,
)
let response = exchange(client, meth, target, headers, request.body, timer) catch {
@http.HttpError::Network(..) if reused && replayable(request) => {
// A parked connection the server had already dropped: replay once
// on a fresh one. `reused` is false on that second attempt.
fresh = true
continue
}
error => raise error
}
let head = response_head(response)
let decoder = if negotiated && has_body(request.http_method, head.status) {
errdefer client.close()
decode_head(head)
} else {
None
}
return {
head,
client,
decoder,
key,
user_agent,
keep_alive: reusable(request.http_method, response),
}
}
}
///|
/// Writes one request as `Write` phases of `timer` (the head, then the body
/// slice by slice, see `send_body`) and reads its response head as the `Head`
/// phase. Each slice is flushed within its phase, the head with the first
/// slice or alone when there is no body; the few bytes that end a
/// chunked body go out with the head's phase. The connection is closed on any
/// failure, including cancellation, which bypasses `catch`.
async fn exchange(
client : @ahttp.Client,
meth : @ahttp.RequestMethod,
target : String,
headers : @ahttp.Headers,
body : @http.Body,
timer : PhaseTimer,
) -> @ahttp.Response raise @http.HttpError {
errdefer client.close()
// A stream is taken before the head goes out, so one that was already sent
// fails without a byte of this request reaching the server.
let outgoing = open_body(body)
defer outgoing.close()
let empty = body.length() is Some(0L)
if empty {
// A head declaring no body goes out whole before any body is read, so an
// empty stream proves it is empty first, and is not read again.
confirm_empty(outgoing, timer)
}
// A stream's first read happens before the head is serialized, since the
// client may flush a large head on its own: whatever a stream checks when
// first read (a multipart body takes every part) fails before any byte.
let first = if !empty && outgoing is Reader(reader) {
Some(timer.run(Write, () => reader.read_some()))
} else {
None
}
timer.run(Write, () => {
client.request(meth, target, extra_headers=headers)
// With a body, the head goes out with its first slice.
if empty {
client.flush()
}
})
if !empty {
send_outgoing(
outgoing,
slice => {
client.write(slice)
client.flush()
},
timer,
buffer_limit=buffered_stream_limit,
first?,
)
// A stream of unknown length that ended at once wrote no slice, so its
// head is still unsent: it goes out in the Write phase, not the Head one.
timer.run(Write, () => client.flush())
}
timer.run(Head, () => client.end_request())
}
///|
fn request_method(meth : String) -> @ahttp.RequestMethod raise @http.HttpError {
match meth.to_lower() {
"get" => Get
"head" => Head
"post" => Post
"put" => Put
"delete" => Delete
"connect" => Connect
"options" => Options
"trace" => Trace
"patch" => Patch
_ => raise @http.HttpError::Protocol("unknown HTTP method: " + meth)
}
}
///|
/// The origin of an absolute `http://` or `https://` URL, and its request
/// target: the path and query, without the fragment.
fn split_url(url : String) -> (Origin, String) raise @http.HttpError {
guard url.find("://") is Some(scheme_end) else {
raise @http.HttpError::Protocol("expected an absolute HTTP URL")
}
let scheme = url[:scheme_end].to_owned().to_lower()
guard scheme == "http" || scheme == "https" else {
raise @http.HttpError::Protocol("unsupported URL scheme: " + scheme)
}
let rest = url[scheme_end + 3:]
let end = rest.find("#").unwrap_or(rest.length())
let rest = rest[:end]
let authority_end = rest
.find("/")
.unwrap_or(rest.length())
.min(rest.find("?").unwrap_or(rest.length()))
let authority = rest[:authority_end]
guard !authority.is_empty() else {
raise @http.HttpError::Protocol("empty URL authority")
}
let tail = rest[authority_end:]
let path = if tail.has_prefix("/") {
tail.to_owned()
} else {
"/" + tail.to_owned()
}
(canonical_origin(scheme, authority.to_owned()), path)
}
///|
/// A request's origin, and the host and port it names, which pick its route.
priv struct Origin {
/// `scheme://host[:port]` in one spelling per server, so the idle pool and
/// the connection caps key every spelling alike: lower-case scheme and
/// host, canonical IPv6 in brackets, no default port.
text : String
https : Bool
/// The host as `@http.Url` keeps it: an IPv6 literal without brackets.
host : String
/// The port, the scheme's default when the URL names none.
port : Int32
}
///|
/// The origin of `scheme://authority`. Userinfo is dropped: it is not part of
/// an origin, and the transport never sends it.
fn canonical_origin(
scheme : String,
authority : String,
) -> Origin raise @http.HttpError {
let url = @http.Url::parse(scheme + "://" + authority + "/") catch {
@http.UrlError::Invalid(reason) =>
raise @http.HttpError::Protocol("invalid URL authority: " + reason)
}
let builder = StringBuilder()
builder.write_string(scheme)
builder.write_string("://")
let host = url.host.unwrap_or("")
if host.contains(":") {
// An IPv6 literal: `Url` keeps the host without its brackets.
builder.write_char('[')
builder.write_string(host)
builder.write_char(']')
} else {
builder.write_string(host)
}
if url.port is Some(port) {
builder.write_char(':')
builder.write_string(port.to_string())
}
let https = scheme == "https"
{
text: builder.to_string(),
https,
host,
port: url.port.unwrap_or(if https { 443 } else { 80 }),
}
}
///|
fn response_head(response : @ahttp.Response) -> @http.ResponseHead {
let headers = @http.Headers::new()
for name, value in response.headers {
headers.append(name.0, value)
}
// async exposes Set-Cookie separately; preserve each parsed cookie as a value.
for cookie in response.cookies {
let text = StringBuilder()
text.write_string(cookie.name + "=" + cookie.value)
if cookie.path is Some(value) {
text.write_string("; Path=" + value)
}
if cookie.expires_raw is Some(value) {
text.write_string("; Expires=" + value)
}
if cookie.max_age is Some(value) {
text.write_string("; Max-Age=" + value.to_string())
}
if cookie.domain is Some(value) {
text.write_string("; Domain=" + value)
}
if cookie.secure {
text.write_string("; Secure")
}
if cookie.http_only {
text.write_string("; HttpOnly")
}
for extension in cookie.extensions {
text.write_string("; " + extension)
}
headers.append("set-cookie", text.to_string())
}
{ status: response.code, headers, }
}