🥮 mooncakes.io
    Sign in

    http-async

    gaato/http Transport over moonbitlang/async. Native, JS, and moonrun Wasm targets.

    http
    transport
    async
    client
    Download zip
    Author
    Version
    0.3.0
    License
    Apache-2.0
    Repository
    Last updated
    21 hours ago
    Downloads
    1K

    Dependencies

    #gaato/http-async

    AsyncTransport and AsyncClock: the gaato/http Transport and Clock over moonbitlang/async, for native, JS, and linear-memory Wasm on moonrun.

    gaato/http-async/mock has VirtualClock, a Clock for async tests whose time moves only when the test advances it: sleeps fire in deadline order, and settle lets the tasks under test run until they go idle between timers, so timing-dependent code runs the same way on every run and target.

    with_timeout_opt(clock, milliseconds, f) is @async.with_timeout_opt with the deadline taken from a Clock, so code that accepts a clock can have its timeouts driven by VirtualClock too.

    The Wasm backend uses MoonBit host APIs for sockets and TLS; it is not a browser or generic WASI transport. When using a moonrun --policy file, allow the destination host and port in net.connect. Loopback tests also need local socket bind permission. Native and Wasm share the HTTP implementation and connection/protocol error classification; JS uses fetch, which connects, sends and waits for the head in one call and does not say how it failed, so its failures before the head are Network(phase=Unknown), never Connect.

    AsyncTransport::new takes timeout_ms (30 s) for a whole buffered exchange, or a stream up to its head, and per-phase limits that default to it: connect_timeout_ms (DNS, TCP, TLS), write_timeout_ms, read_timeout_ms (the head, then each body read; it alone bounds a stream's reads) and pool_timeout_ms. A Timeout names the phase it cut short. max_connections and max_connections_per_origin cap the connections in use; a request over a cap waits its turn until pool_timeout_ms. Without them, the default, requests never wait.

    Request bodies go out in writes of at most 64 KiB, each bounded by write_timeout_ms. A body in memory, or a stream body declared with a length (Request::body_stream(stream, length=...)), is sent with Content-Length; a stream without one is sent chunked, read as it is written. On JS, moonbitlang/async's client 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.

    AsyncTransport keeps a small keep-alive pool per origin (max_idle_per_origin, idle_max_ms) and replays a request once on a fresh connection when a reused one fails before the response head, but only for idempotent methods or requests carrying an Idempotency-Key, and never with a stream body it already read. Call close() to drop parked connections. On native and Wasm the transport sends Accept-Encoding: gzip, deflate, br and decodes the body as it reads it, unless the request sets Accept-Encoding itself; a coding it does not know is left on the body. Each br body may hold up to 16 MiB of window, and only as much as it has decoded. max_body_bytes caps a buffered response body after decoding; a longer body closes the connection and fails with Protocol, while send_stream leaves the limit to its reader. It is unbounded by default, so set it: AsyncTransport::new(max_body_bytes=64 * 1024 * 1024) is ample for API JSON, and the SDKs built on gaato/sdk-runtime take the transport from you and cannot set it themselves.

    On native and Wasm, AsyncTransport::new(proxy="http://user:pass@proxy:3128") sends every request through an HTTP proxy (https:// proxies too): http:// requests are forwarded, https:// ones tunnelled with CONNECT. Without proxy, HTTP_PROXY, HTTPS_PROXY, ALL_PROXY and NO_PROXY are read when the transport is made, as curl reads them; trust_env=false turns that off. A refused tunnel raises HttpError::Proxy with the proxy's status. On JS, fetch follows the platform's proxy settings, and a transport given proxy refuses its requests.

    In 0.1.2, a request's User-Agent is set when its connection is opened. Connections are reused only for requests with the same User-Agent; requests without one retain moonbitlang/async's default.

    Unofficial and experimental. Source, issues and design notes: https://github.com/gaato/mbt-sdk

    Known limitation: moonbitlang/async 0.22.5 cannot dial an IPv6 literal host (http://[::1]:8080/, as an origin or a proxy): it passes the bracketed address to name resolution, which fails with Connect. Use a host name until that is fixed upstream.

    AsyncClock

    source
    pub struct AsyncClock {
    }

    A production clock backed by moonbitlang/async timers and wall time.
    impl Clock for AsyncClock

    AsyncClock::new

    source
    fn AsyncClock::new() -> AsyncClock

    Creates an async clock.

    test {
      let _ = @adapter.AsyncClock::new()
    }

    AsyncClock::now_unix_ms

    source
    fn AsyncClock::now_unix_ms(_self : AsyncClock) -> Int64

    Explicit clock methods backed by moonbitlang/async.

    AsyncClock::sleep

    source
    async fn AsyncClock::sleep(_self : AsyncClock, milliseconds : Int) -> Unit noraise

    Explicit clock methods backed by moonbitlang/async.

    AsyncTransport

    source
    pub struct AsyncTransport {
      // private fields
    }

    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.

    AsyncTransport::close

    source
    fn AsyncTransport::close(self : AsyncTransport) -> Unit

    Closes every idle connection and refuses further requests, and those waiting for a connection slot, with Connect("transport closed"). In-flight requests finish and their connections are closed instead of parked.

    test {
      let transport = @adapter.AsyncTransport::new()
      transport.close()
      assert_eq(transport.idle_connections(), 0)
    }

    AsyncTransport::idle_connections

    source
    fn AsyncTransport::idle_connections(self : AsyncTransport) -> Int

    The number of parked keep-alive connections across all origins.

    AsyncTransport::new

    source
    fn AsyncTransport::new(timeout_ms? : Int, connect_timeout_ms? : Int, write_timeout_ms? : Int, read_timeout_ms? : Int, pool_timeout_ms? : Int, max_connections? : Int, max_connections_per_origin? : Int, max_idle_per_origin? : Int, idle_max_ms? : Int, max_body_bytes? : Int, proxy? : String, trust_env? : Bool) -> AsyncTransport

    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.

    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)
    }

    AsyncTransport::send

    source

    Explicit transport methods for buffered and streaming requests.

    AsyncTransport::send_stream

    source

    Explicit transport methods for buffered and streaming requests.

    DeadlineTransport

    source
    pub struct DeadlineTransport {
      // private fields
    }

    A per-exchange deadline: the whole buffered exchange, or only the response head for a streaming exchange. Expiry raises Timeout(phase=Unknown,ms=timeout_ms), since the deadline cannot see which phase the inner transport was in; the inner transport's own failures pass through.

    DeadlineTransport::new

    source

    Wraps a transport with a deadline; nonpositive deadlines expire immediately.

    DeadlineTransport::send

    source

    Explicit transport methods with per-exchange deadlines.

    DeadlineTransport::send_stream

    source

    Explicit transport methods with per-exchange deadlines.

    SemaphoreGate

    source
    pub struct SemaphoreGate {
      // private fields
    }

    An admission gate backed by a cancellable async semaphore.

    SemaphoreGate::acquire

    source
    async fn SemaphoreGate::acquire(self : SemaphoreGate) -> Unit noraise

    Explicit gate methods backed by the semaphore.

    SemaphoreGate::new

    source
    fn SemaphoreGate::new(permits? : Int) -> SemaphoreGate

    Creates a gate with a positive number of permits (one by default).

    SemaphoreGate::release

    source
    fn SemaphoreGate::release(self : SemaphoreGate) -> Unit

    Explicit gate methods backed by the semaphore.

    with_timeout_opt

    source
    async fn[X] with_timeout_opt(clock : &
    Clock    , milliseconds : Int, f : async () -> X) -> X?

    Runs f, giving up after milliseconds on clock: None when the time runs out first, in which case f is cancelled. Errors from f propagate. The clock-driven form of @async.with_timeout_opt, so a test can expire the deadline by advancing a virtual clock instead of waiting for it.

    A nonpositive milliseconds has already expired: f still starts, so a result it produces without waiting is returned, but anything it has to wait for gives up.

    async test {
      let clock = @adapter.AsyncClock::new()
      assert_eq(@adapter.with_timeout_opt(clock, 1000, () => 7), Some(7))
    }

    Source Files

    Powered by MoonBit

    Site sourceReport issuePackagesBuild queueSkillsStatistics

    © 2026 mooncakes.io