| Plan | body | The Client's reader |
|---|---|---|
| Buffered | Exchange(max_bytes = max_body_bytes): the deadline covers the body, and the backend admits at most that many wire bytes | keeps at most max_body_bytes decoded bytes |
| Streaming, from_head included | Stream(idle_ms = read_idle_ms): the deadline ends at the head, each read gets the idle limit | a buffered branch the head selects keeps max_body_bytes and the rest of the attempt deadline, enforced with the Client's Timer |
///|
test "retry a 503 and read JSON" {
// A backend that is busy once, then answers.
let mut calls = 0
let backend = @client.from_fn((_, _) => {
calls += 1
if calls == 1 {
@http.Response::new(503)
} else {
@http.Response::new(200, body=Bytes(b"{\"id\":7}"))
}
})
let client = @client.Client::new(
backend,
@sim.FakeClock::new(),
base_url="https://api.example/v1",
retry=@client.RetryPolicy::conservative(),
)
let plan : @client.ClientRequest[Result[Json, @client.ResponseFailure]] = @client.get(
"/items/7",
).response(@client.as_json_value())
let reply = @driver.run_sync(() => client.send(plan))
assert_eq(reply.body, Ok({ "id": 7 }))
assert_eq(reply.request.url, "https://api.example/v1/items/7")
assert_eq(calls, 2)
}pub trait ResponseMode {
fn fresh() -> Self
fn may_stream(Self) -> Bool
}pub struct Admission {
// private fields
}pub(all) struct Attempt {
request : RequestHead
attributes : Attributes
bucket : String
attempt : Int
status : Int
headers : Headers
duration_ms : Int64
error : Error?
decision : Decision?
}pub(all) enum AttemptResult[T] {
Buffered(Reply[T])
StreamReady(ResponseHead)
TransportFailure(TransportError)
}pub(all) struct Buffered {
}impl ResponseMode for Bufferedpub struct Client {
// private fields
}fn Client::new(backend : &Backend, clock : &Clock, base_url? : String, headers? : Headers, timeout_ms? : Int, max_body_bytes? : Int, read_idle_ms? : Int, timer? : &Timer, retry? : RetryPolicy, limiter? : &RateLimiter, total_ms? : Int, observer? : (Attempt) -> Unit, random? : () -> Double) -> Clientasync fn[T, M : ResponseMode] Client::send(self : Client, request : RequestPlan[T, M], decide_retry? : (AttemptResult[T], RetryContext) -> Decision) -> Reply[T]pub struct FollowRedirects {
// private fields
}impl Backend for FollowRedirectsasync fn send(self : FollowRedirects, request : Request, options : SendOptions) -> Response raise TransportErrorfn FollowRedirects::new(inner : &Backend, config? : RedirectConfig, on_redirect? : (RedirectHop) -> Unit) -> FollowRedirectsasync fn FollowRedirects::send(self : FollowRedirects, request : Request, options : SendOptions) -> Response raise TransportErrorpub struct NoLimiter {
// private fields
}impl RateLimiter for NoLimiterpub struct Paginator[T, E] {
// private fields
}test {
let config = {
..@client.RedirectConfig::default(),
credential_headers: [
"authorization", "proxy-authorization", "cookie", "x-api-key",
],
}
assert_eq(config.max_redirects, 10)
assert_false(config.allow_downgrade)
}pub struct RedirectHop {
request : RequestHead
head : ResponseHead
next : RequestHead
} derive(Eq, Debug)pub(all) struct Reply[T] {
status : Int
headers : Headers
body : T
request : RequestHead
attributes : Attributes
}pub struct RequestPlan[T, M] {
// private fields
}fn[T, A, M] RequestPlan::attribute(self : RequestPlan[T, M], key : AttributeKey[A], value : A) -> RequestPlan[T, M]fn[T, M] RequestPlan::follow_redirects(self : RequestPlan[T, M], enabled : Bool) -> RequestPlan[T, M]fn[T, M] RequestPlan::header(self : RequestPlan[T, M], name : String, value : String) -> RequestPlan[T, M]fn[T, U, M, N] RequestPlan::response(self : RequestPlan[T, M], response_as : ResponsePlan[U, N]) -> RequestPlan[U, N]async fn[T, M : ResponseMode] RequestPlan::send(self : RequestPlan[T, M], client : Client, decide_retry? : (AttemptResult[T], RetryContext) -> Decision) -> Reply[T]fn[T] RequestPlan::to_curl(self : RequestPlan[T, Buffered], sensitive_headers? : Array[String], include_body? : Bool) -> Stringpub(all) enum ResponseFailure {
UnexpectedStatus(ResponseHead, body~ : Bytes)
Deserialization(ResponseHead, message~ : String, body~ : Bytes)
} derive(Eq, Debug)pub struct ResponsePlan[T, M] {
// private fields
}fn[A, B, E, M] ResponsePlan::map_ok(self : ResponsePlan[Result[A, E], M], f : (A) -> B raise) -> ResponsePlan[Result[B, E], M]fn[A, B, M] ResponsePlan::map_with_head(self : ResponsePlan[A, M], f : (A, ResponseHead) -> B raise) -> ResponsePlan[B, M]fn[A, T] ResponsePlan::prepared_stream(decode : (ResponseHead, Bytes) -> Result[A, T] raise, use_ : async (A) -> T) -> ResponsePlan[T, Streaming]fn[T] ResponsePlan::stream(use_ : async (ResponseHead, &BodyStream, Int) -> T) -> ResponsePlan[Result[T, ResponseFailure], Streaming]pub(all) struct RetryContext {
request : RequestHead
idempotent : Bool
bucket : String
attempt : Int
random : Double
now_unix_ms : Int64
remaining_ms : Int64?
replayable : Bool
default_decision : Decision
}fn RetryContext::capture(request : Request, attempt? : Int, bucket? : String, random? : Double, now_unix_ms? : Int64, remaining_ms? : Int64, replayable? : Bool) -> RetryContextfn RetryContext::new(request : RequestHead, idempotent~ : Bool, replayable~ : Bool, attempt? : Int, bucket? : String, random? : Double, now_unix_ms? : Int64, remaining_ms? : Int64, default_decision? : Decision) -> RetryContextfn RetryPolicy::decide(self : RetryPolicy, outcome : AttemptOutcome, context : RetryContext) -> Decisionpub(all) struct Streaming {
}impl ResponseMode for Streamingfn[A, B] as_both(left : ResponsePlan[A, Buffered], right : ResponsePlan[B, Buffered]) -> ResponsePlan[(A, B), Buffered]fn as_form(max_pairs? : Int) -> ResponsePlan[Result[Array[(String, String)], ResponseFailure], Buffered]fn as_lines(receive : async (String) -> Unit, max_line_bytes? : Int) -> ResponsePlan[Result[Unit, ResponseFailure], Streaming]fn as_sse(receive : async (SseEvent) -> Unit, max_event_bytes? : Int, stop? : (SseEvent) -> Bool) -> ResponsePlan[Result[Unit, ResponseFailure], Streaming]fn[T] as_stream(use_ : async (ResponseHead, &BodyStream) -> T) -> ResponsePlan[Result[T, ResponseFailure], Streaming]fn from_fn(send : async (Request, SendOptions) -> Response raise TransportError, on_close? : () -> Unit) -> &Backendfn[T, M : ResponseMode] from_head(choose : (ResponseHead) -> ResponsePlan[T, M]) -> ResponsePlan[T, M]let max_error_body : Intfn parse_retry_after(value : String, now_unix_ms? : Int64) -> Int?test {
assert_eq(@retry.parse_retry_after("1.2349"), Some(1234))
}Install
Download zipHTTP for MoonBit in the style of sttp, on top of gaato/http: a client with retries, rate limits and deadlines, server-sent events, multipart, cookies, endpoints and OpenAPI. Runs on every backend.