Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522//! HTTP on siv's event loop: libcurl's multi interface, with its sockets//! watched by siv.appWatchFd and its timeouts kept by siv.timerStart, so//! requests run on the app's thread and never block it. Requests to one//! host share HTTP/2 connections, ten in flight at most, as Google turns//! away a user's requests past some number at once; the rest wait their//! turn in curl. A reply that says to slow down is retried, later. A//! reply that goes on, such as server-sent events, is heard as it comes.//!//! libcurl stands in until baskingspot/http and baskingspot/tls build on//! Zig 0.17: they are sans-I/O too, and would sit behind the same watches.//! The declarations below are the few parts of curl.h it needs, written//! out because Zig 0.17 has no @cImport; their values are libcurl's ABI.
const std = @import("std");const siv = @import("siv");const ascii = std.ascii;const mem = std.mem;const testing = std.testing;const Allocator = std.mem.Allocator;const ArrayList = std.ArrayList;const Io = std.Io;const log = std.log.scoped(.http);
const Http = @This();
gpa: Allocator,app: siv.App,multi: *c.Multi,/// When curl wants to be called with no socket ready; null when it does/// not.timer: ?siv.Timer,/// Requests in flight, to free if the client goes first.transfers: ArrayList(*Transfer),
pub const Request = struct { method: Method, url: []const u8, /// Whole header lines, such as "authorization: Bearer …". headers: []const []const u8, /// What a POST or a PUT sends. body: ?[]const u8,};
pub const Method = enum { get, post, put, delete };
pub const Response = struct { status: u16, /// Freed once the callback returns. body: []const u8,};
pub const Error = error{ HttpFailed, OutOfMemory };
/// Hears how a request went, on the app's thread.pub const Done = *const fn (context: *anyopaque, result: Error!Response) void;
/// Hears what has come of a reply that goes on, on the app's thread.pub const Heard = *const fn (context: *anyopaque, bytes: []const u8) void;
pub fn init(gpa: Allocator, app: siv.App) !*Http { try globalInit(); const self = try gpa.create(Http); errdefer gpa.destroy(self); const multi = c.curl_multi_init() orelse return error.HttpFailed; self.* = .{ .gpa = gpa, .app = app, .multi = multi, .timer = null, .transfers = .empty }; _ = c.curl_multi_setopt(multi, c.mopt_socketfunction, &onSocket); _ = c.curl_multi_setopt(multi, c.mopt_socketdata, self); _ = c.curl_multi_setopt(multi, c.mopt_timerfunction, &onTimer); _ = c.curl_multi_setopt(multi, c.mopt_timerdata, self); _ = c.curl_multi_setopt(multi, c.mopt_max_host_connections, @as(c_long, 2)); _ = c.curl_multi_setopt(multi, c.mopt_max_concurrent_streams, @as(c_long, 5)); return self;}
/// Drops the requests in flight without calling their callbacks.pub fn deinit(self: *Http) void { if (self.timer) |t| siv.timerCancel(t); for (self.transfers.items) |t| { if (t.retry) |r| siv.timerCancel(r) else _ = c.curl_multi_remove_handle(self.multi, t.easy); t.destroy(); } self.transfers.deinit(self.gpa); _ = c.curl_multi_cleanup(self.multi); self.gpa.destroy(self);}
/// Starts a request; `done` hears how it went, never before this returns.pub fn send(self: *Http, request: Request, context: *anyopaque, done: Done) Error!void { const t = try Transfer.create(self.gpa, request); errdefer t.destroy(); try self.start(t, context, done);}
/// Starts a GET whose reply goes on, such as server-sent events: `heard`/// hears a 2xx reply's bytes as they come, and `done` how it ended, with/// any other reply's body. One silent for `silent_s` seconds fails.pub fn listen(self: *Http, request: Request, silent_s: u32, context: *anyopaque, heard: Heard, done: Done) Error!void { const t = try Transfer.create(self.gpa, request); errdefer t.destroy(); t.heard = heard; // curl's speed is over the last few seconds, so a byte now and then, // such as a ping, keeps it above 1 B/s for long enough to reset this. try ok(c.curl_easy_setopt(t.easy, c.opt_low_speed_limit, @as(c_long, 1))); try ok(c.curl_easy_setopt(t.easy, c.opt_low_speed_time, @as(c_long, silent_s))); // A compressed stream may hold back what has come. try ok(c.curl_easy_setopt(t.easy, c.opt_accept_encoding, @as(?[*:0]const u8, null))); try self.start(t, context, done);}
fn start(self: *Http, t: *Transfer, context: *anyopaque, done: Done) Error!void { t.context = context; t.done = done; t.http = self; try self.transfers.append(self.gpa, t); errdefer _ = self.transfers.pop(); if (c.curl_multi_add_handle(self.multi, t.easy) != 0) return error.HttpFailed;}
/// Makes a request and waits for it, for the command line, where there/// is no event loop. `arena` holds the body.pub fn fetch(arena: Allocator, request: Request) Error!Response { try globalInit(); const t = try Transfer.create(arena, request); defer t.destroy(); const code = c.curl_easy_perform(t.easy); if (code != 0) { log.err("{s}: {s}", .{ request.url, t.reason(code) }); return error.HttpFailed; } return .{ .status = t.status(), .body = try arena.dupe(u8, t.body.items) };}
fn globalInit() Error!void { // curl_global_init is not thread-safe; it is called before any thread // makes a request, and again harmlessly. if (c.curl_global_init(c.global_default) != 0) return error.HttpFailed;}
/// The largest reply taken: Gmail's 25 MB attachment limit, in base64/// and JSON.const max_body = 40 << 20;
/// How long to wait before each retry of a request told to slow down:/// about a minute in all, as Google's quotas are per minute.const backoff_ms = [_]u32{ 1000, 2000, 4000, 8000, 16_000, 32_000 };
/// Whether a reply says to slow down: a 429, or a 403 for a rate limit/// or a quota, as Google's APIs answer them.fn slowDown(status: u16, body: []const u8) bool { return status == 429 or (status == 403 and (ascii.findIgnoreCase(body, "ratelimitexceeded") != null or ascii.findIgnoreCase(body, "quota exceeded") != null));}
/// One request and what has come back of it.const Transfer = struct { gpa: Allocator, /// The client sending it; unset for `fetch`. http: *Http, easy: *c.Easy, headers: ?*c.Slist, body: ArrayList(u8), context: *anyopaque, done: Done, /// Set for a reply that goes on, which hears its body as it comes. heard: ?Heard, error_buf: [c.error_size]u8, /// Retries so far. tries: u8, /// Set while it waits to be retried, off curl's books. retry: ?siv.Timer,
fn create(gpa: Allocator, request: Request) Error!*Transfer { const t = try gpa.create(Transfer); errdefer gpa.destroy(t); const easy = c.curl_easy_init() orelse return error.HttpFailed; t.* = .{ .gpa = gpa, .http = undefined, .easy = easy, .headers = null, .body = .empty, .context = undefined, .done = undefined, .heard = null, .error_buf = @splat(0), .tries = 0, .retry = null, }; errdefer t.destroy(); // curl copies strings it is given, but wants them NUL-terminated. var z: std.heap.ArenaAllocator = .init(gpa); defer z.deinit(); const url = try z.allocator().dupeSentinel(u8, request.url, 0); for (request.headers) |h| { t.headers = c.curl_slist_append(t.headers, try z.allocator().dupeSentinel(u8, h, 0)) orelse return error.OutOfMemory; } try ok(c.curl_easy_setopt(easy, c.opt_url, url.ptr)); try ok(c.curl_easy_setopt(easy, c.opt_private, t)); try ok(c.curl_easy_setopt(easy, c.opt_writefunction, &onData)); try ok(c.curl_easy_setopt(easy, c.opt_writedata, t)); try ok(c.curl_easy_setopt(easy, c.opt_errorbuffer, &t.error_buf)); try ok(c.curl_easy_setopt(easy, c.opt_nosignal, @as(c_long, 1))); try ok(c.curl_easy_setopt(easy, c.opt_accept_encoding, "".ptr)); try ok(c.curl_easy_setopt(easy, c.opt_useragent, "parcel".ptr)); try ok(c.curl_easy_setopt(easy, c.opt_connecttimeout_ms, @as(c_long, 15_000))); // Wait for a connection that can multiplex rather than open more. try ok(c.curl_easy_setopt(easy, c.opt_pipewait, @as(c_long, 1))); // Images a message shows come from anywhere, often by way of a // redirect: web addresses only, a few hops, and nothing huge. try ok(c.curl_easy_setopt(easy, c.opt_protocols_str, "http,https".ptr)); try ok(c.curl_easy_setopt(easy, c.opt_redir_protocols_str, "http,https".ptr)); try ok(c.curl_easy_setopt(easy, c.opt_followlocation, @as(c_long, 1))); try ok(c.curl_easy_setopt(easy, c.opt_maxredirs, @as(c_long, 5))); try ok(c.curl_easy_setopt(easy, c.opt_maxfilesize_large, @as(i64, max_body))); if (t.headers) |h| try ok(c.curl_easy_setopt(easy, c.opt_httpheader, h)); if (request.method == .post or request.method == .put) { const body = request.body orelse ""; try ok(c.curl_easy_setopt(easy, c.opt_postfieldsize, @as(c_long, @intCast(body.len)))); try ok(c.curl_easy_setopt(easy, c.opt_copypostfields, body.ptr)); } // A POST, under another name. if (request.method == .put) try ok(c.curl_easy_setopt(easy, c.opt_customrequest, "PUT".ptr)); if (request.method == .delete) try ok(c.curl_easy_setopt(easy, c.opt_customrequest, "DELETE".ptr)); return t; }
fn destroy(t: *Transfer) void { c.curl_easy_cleanup(t.easy); c.curl_slist_free_all(t.headers); t.body.deinit(t.gpa); t.gpa.destroy(t); }
/// Hands the request back to curl after a wait, if it has tries left. fn retryLater(t: *Transfer) bool { if (t.tries == backoff_ms.len) return false; const delay = backoff_ms[t.tries]; t.retry = siv.timerStart(t.http.app, delay, onRetry, null, t) catch |err| { log.err("setting a retry's timer: {t}", .{err}); return false; }; t.tries += 1; log.info("asked to slow down; trying again in {d} s", .{delay / 1000}); return true; }
fn onRetry(user_data: ?*anyopaque) callconv(.c) void { const t: *Transfer = @ptrCast(@alignCast(user_data)); t.retry = null; t.body.clearRetainingCapacity(); if (c.curl_multi_add_handle(t.http.multi, t.easy) == 0) return; // As if curl had failed it. const http = t.http; if (mem.findScalar(*Transfer, http.transfers.items, t)) |i| _ = http.transfers.swapRemove(i); defer t.destroy(); t.done(t.context, error.HttpFailed); }
fn status(t: *Transfer) u16 { var code: c_long = 0; _ = c.curl_easy_getinfo(t.easy, c.info_response_code, &code); return std.math.cast(u16, code) orelse 0; }
fn reason(t: *const Transfer, code: c_int) []const u8 { const detail = mem.sliceTo(&t.error_buf, 0); return if (detail.len > 0) detail else mem.span(c.curl_easy_strerror(code)); }
fn onData(ptr: [*]const u8, size: usize, count: usize, userdata: *anyopaque) callconv(.c) usize { const t: *Transfer = @ptrCast(@alignCast(userdata)); const n = size * count; // Fewer bytes than given makes curl fail the transfer. if (t.body.items.len + n > max_body) return 0; t.body.appendSlice(t.gpa, ptr[0..n]) catch return 0; return n; }};
fn ok(code: c_int) Error!void { if (code != 0) return error.HttpFailed;}
/// A socket curl uses, and siv's watch on it.const Socket = struct { http: *Http, fd: c_int, watch: siv.Watch,
fn onReady(user_data: ?*anyopaque, _: siv.Watch, events: siv.WatchFlags) callconv(.c) void { const s: *Socket = @ptrCast(@alignCast(user_data)); var mask: c_int = 0; if (events.readable) mask |= c.cselect_in; if (events.writable) mask |= c.cselect_out; if (events.@"error") mask |= c.cselect_err; // This may remove the watch, and free `s`: keep what is needed. const http = s.http; http.act(s.fd, mask); }
fn destroy(user_data: ?*anyopaque) callconv(.c) void { const s: *Socket = @ptrCast(@alignCast(user_data)); s.http.gpa.destroy(s); }};
/// curl says which sockets to watch, and for what.fn onSocket(_: *c.Easy, fd: c_int, what: c_int, userp: *anyopaque, socketp: ?*anyopaque) callconv(.c) c_int { const self: *Http = @ptrCast(@alignCast(userp)); const existing: ?*Socket = @ptrCast(@alignCast(socketp)); if (what == c.poll_remove) { if (existing) |s| siv.watchDestroy(s.watch); return 0; } const events: siv.WatchFlags = .{ .readable = what & c.poll_in != 0, .writable = what & c.poll_out != 0 }; if (existing) |s| { siv.watchSetEvents(s.watch, events) catch |err| log.err("watching a socket: {t}", .{err}); return 0; } const s = self.gpa.create(Socket) catch return -1; s.* = .{ .http = self, .fd = fd, .watch = undefined }; s.watch = siv.appWatchFd(self.app, fd, events, Socket.onReady, Socket.destroy, s) catch |err| { log.err("watching a socket: {t}", .{err}); self.gpa.destroy(s); return -1; }; _ = c.curl_multi_assign(self.multi, fd, s); return 0;}
/// curl says when to call it back if no socket is ready first.fn onTimer(_: *c.Multi, timeout_ms: c_long, userp: *anyopaque) callconv(.c) c_int { const self: *Http = @ptrCast(@alignCast(userp)); if (self.timer) |t| siv.timerCancel(t); self.timer = null; if (timeout_ms < 0) return 0; const delay: u32 = @intCast(@min(timeout_ms, std.math.maxInt(u32))); self.timer = siv.timerStart(self.app, delay, onTimeout, null, self) catch |err| { log.err("setting curl's timer: {t}", .{err}); return -1; }; return 0;}
fn onTimeout(user_data: ?*anyopaque) callconv(.c) void { const self: *Http = @ptrCast(@alignCast(user_data)); self.timer = null; self.act(c.socket_timeout, 0);}
/// Lets curl work on a ready socket, or on its timeout, then hands out/// the requests that finished.fn act(self: *Http, fd: c_int, mask: c_int) void { var running: c_int = 0; _ = c.curl_multi_socket_action(self.multi, fd, mask, &running); // Out of curl's callbacks, so what hears it may send requests. var at: usize = 0; while (at < self.transfers.items.len) : (at += 1) { const t = self.transfers.items[at]; const heard = t.heard orelse continue; if (t.body.items.len == 0 or t.status() / 100 != 2) continue; heard(t.context, t.body.items); t.body.clearRetainingCapacity(); } var left: c_int = 0; while (c.curl_multi_info_read(self.multi, &left)) |msg| { if (msg.msg != c.msg_done) continue; var private: ?*anyopaque = null; _ = c.curl_easy_getinfo(msg.easy, c.info_private, &private); const t: *Transfer = @ptrCast(@alignCast(private.?)); const code = msg.data.result; _ = c.curl_multi_remove_handle(self.multi, t.easy); if (code == 0 and slowDown(t.status(), t.body.items) and t.retryLater()) continue; if (mem.findScalar(*Transfer, self.transfers.items, t)) |i| _ = self.transfers.swapRemove(i); defer t.destroy(); if (code != 0) { log.warn("{s}", .{t.reason(code)}); t.done(t.context, error.HttpFailed); } else { t.done(t.context, .{ .status = t.status(), .body = t.body.items }); } }}
/// The parts of curl.h used here.const c = struct { const Easy = opaque {}; const Multi = opaque {}; const Slist = opaque {}; const Msg = extern struct { msg: c_int, easy: *Easy, data: extern union { whatever: ?*anyopaque, result: c_int }, };
const global_default: c_long = 3; const error_size = 256;
const opt_writedata: c_int = 10001; const opt_url: c_int = 10002; const opt_errorbuffer: c_int = 10010; const opt_writefunction: c_int = 20011; const opt_useragent: c_int = 10018; const opt_httpheader: c_int = 10023; const opt_customrequest: c_int = 10036; const opt_postfieldsize: c_int = 60; const opt_nosignal: c_int = 99; const opt_accept_encoding: c_int = 10102; const opt_private: c_int = 10103; const opt_connecttimeout_ms: c_int = 156; const opt_copypostfields: c_int = 10165; const opt_pipewait: c_int = 237; const opt_followlocation: c_int = 52; const opt_maxredirs: c_int = 68; const opt_low_speed_limit: c_int = 19; const opt_low_speed_time: c_int = 20; const opt_maxfilesize_large: c_int = 30117; const opt_protocols_str: c_int = 10318; const opt_redir_protocols_str: c_int = 10319;
const mopt_socketfunction: c_int = 20001; const mopt_socketdata: c_int = 10002; const mopt_timerfunction: c_int = 20004; const mopt_timerdata: c_int = 10005; const mopt_max_host_connections: c_int = 7; const mopt_max_concurrent_streams: c_int = 16;
const info_response_code: c_int = 0x200002; const info_private: c_int = 0x100015;
const socket_timeout: c_int = -1; const cselect_in: c_int = 1; const cselect_out: c_int = 2; const cselect_err: c_int = 4; const poll_in: c_int = 1; const poll_out: c_int = 2; const poll_remove: c_int = 4; const msg_done: c_int = 1;
extern fn curl_global_init(flags: c_long) c_int; extern fn curl_easy_init() ?*Easy; extern fn curl_easy_cleanup(easy: *Easy) void; extern fn curl_easy_setopt(easy: *Easy, option: c_int, ...) c_int; extern fn curl_easy_getinfo(easy: *Easy, info: c_int, ...) c_int; extern fn curl_easy_perform(easy: *Easy) c_int; extern fn curl_easy_strerror(code: c_int) [*:0]const u8; extern fn curl_slist_append(list: ?*Slist, string: [*:0]const u8) ?*Slist; extern fn curl_slist_free_all(list: ?*Slist) void; extern fn curl_multi_init() ?*Multi; extern fn curl_multi_cleanup(multi: *Multi) c_int; extern fn curl_multi_setopt(multi: *Multi, option: c_int, ...) c_int; extern fn curl_multi_add_handle(multi: *Multi, easy: *Easy) c_int; extern fn curl_multi_remove_handle(multi: *Multi, easy: *Easy) c_int; extern fn curl_multi_assign(multi: *Multi, fd: c_int, data: ?*anyopaque) c_int; extern fn curl_multi_socket_action(multi: *Multi, fd: c_int, mask: c_int, running: *c_int) c_int; extern fn curl_multi_info_read(multi: *Multi, left: *c_int) ?*Msg;};
test "a request told to slow down is sent again, and its callback hears only the answer" { var address: Io.net.IpAddress = .{ .ip4 = .loopback(0) }; var server = try address.listen(testing.io, .{}); defer server.deinit(testing.io); var serving = try testing.io.concurrent(slowThenFine, .{ &server, testing.io }); defer serving.cancel(testing.io);
const app = try siv.appCreate(.{ .backend = .headless, .flags = .{ .no_system_fonts = true, .no_system_icons = true, .reduced_motion = true }, }); defer siv.appDestroy(app); const http = try init(testing.allocator, app); defer http.deinit(); var url_buf: [64]u8 = undefined; const url = try mem.print(&url_buf, "http://127.0.0.1:{d}/", .{server.socket.address.getPort()}); const Answer = struct { calls: u32 = 0, status: u16 = 0, body: [16]u8 = undefined, len: usize = 0 }; var heard: Answer = .{}; try http.send(.{ .method = .get, .url = url, .headers = &.{}, .body = null }, &heard, struct { fn done(context: *anyopaque, result: Error!Response) void { const h: *Answer = @ptrCast(@alignCast(context)); h.calls += 1; const r = result catch return; h.status = r.status; h.len = @min(r.body.len, h.body.len); @memcpy(h.body[0..h.len], r.body[0..h.len]); } }.done); var waited: u32 = 0; while (heard.calls == 0 and waited < 50) : (waited += 1) try siv.appDispatch(app, 100); try testing.expectEqual(1, heard.calls); try testing.expectEqual(200, heard.status); try testing.expectEqualStrings("fine", heard.body[0..heard.len]);}
/// Answers its first request with Gmail's quota refusal, then "fine".fn slowThenFine(server: *Io.net.Server, io: Io) void { var answered: u32 = 0; while (true) { const stream = server.accept(io) catch |err| switch (err) { error.Canceled => return, else => continue, }; defer stream.close(io); var read_buf: [4096]u8 = undefined; var write_buf: [4096]u8 = undefined; var reader = stream.reader(io, &read_buf); var writer = stream.writer(io, &write_buf); var server_http: std.http.Server = .init(&reader.interface, &writer.interface); while (true) { var request = server_http.receiveHead() catch break; const refused = \\{"error":{"code":403,"message":"Quota exceeded for quota metric 'Total Query Cost'", \\"errors":[{"reason":"rateLimitExceeded"}]}} ; const first = answered == 0; answered += 1; request.respond(if (first) refused else "fine", .{ .status = if (first) .forbidden else .ok }) catch break; } }}