Something went wrong. Try again.
A SpaceTraders Agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446const std = @import("std");const HTTPClient = std.http.Client;const Io = std.Io;const json = std.json;
const buildin = @import("builtin");const models = @import("models.zig");
const log = std.log.scoped(.SpaceTraders);
const Semaphore = struct { mutex: Io.Mutex = .init, cond: Io.Condition = .init, /// It is OK to initialise this field to any value. permits: u64 = 0,
pub fn post(sem: *Semaphore, io: Io) void { sem.mutex.lockUncancelable(io); defer sem.mutex.unlock(io);
sem.permits += 1; sem.cond.signal(io); }
pub fn set(sem: *Semaphore, io: Io, permits: u64) void { sem.mutex.lockUncancelable(io); defer sem.mutex.unlock(io);
sem.permits = permits; sem.cond.signal(io); }
pub fn wait(sem: *Semaphore, io: Io) !void { sem.mutex.lockUncancelable(io); defer sem.mutex.unlock(io);
while (sem.permits == 0) try sem.cond.wait(io, &sem.mutex);
sem.permits -= 1; if (sem.permits > 0) sem.cond.signal(io); }
pub fn available(sem: *Semaphore, io: Io) u64 { sem.mutex.lockUncancelable(io); defer sem.mutex.unlock(io);
return sem.permits; }};
pub const Limiter = struct { points: u64, duration: i64, time: ?Io.Timestamp,
mutex: Io.Mutex = .init, semaphor: Semaphore,
pub fn init(opts: struct { points: u54 = 2, duration: i64 = 1000 }) Limiter { return .{ .points = opts.points, .duration = opts.duration, .time = null, .semaphor = Semaphore{ .permits = opts.points }, }; }
pub fn checkReset(l: *Limiter, io: Io) bool { l.mutex.lock(io) catch return false; defer l.mutex.unlock(io);
if (l.time) |t| { const dur = t.durationTo(Io.Clock.now(.real, io)); if (dur.toSeconds() > 0) { l.semaphor.set(io, l.points); l.time = null; return true; } }
return false; }
pub fn aquire(l: *Limiter, io: Io) !void { try l.mutex.lock(io); defer l.mutex.unlock(io);
if (l.time == null) { const now = Io.Clock.now(.real, io); l.time = now.addDuration(.fromMilliseconds(l.duration)); }
return l.semaphor.wait(io); }
pub fn timeToReset(l: *Limiter, io: Io) Io.Duration { if (l.time) |t| { return t.durationTo(Io.Clock.now(.real, io)); } return .zero; }
pub fn available(l: *Limiter, io: Io) bool { return l.semaphor.available(io) > 0; }};
pub const BurstyLimiter = struct { static: Limiter, burst: Limiter,
pub fn wait(bl: *BurstyLimiter, io: Io) !void { _ = bl.static.checkReset(io); _ = bl.burst.checkReset(io);
if (!bl.static.available(io)) { if (bl.burst.available(io)) { log.debug("Using Burst", .{}); try bl.burst.aquire(io); return; } else { log.warn("No request available, waiting", .{});
var static = bl.static.checkReset(io); var burst = bl.burst.checkReset(io);
while (!static and !burst and !bl.static.available(io)) { try io.sleep(bl.static.timeToReset(io), .real); static = bl.static.checkReset(io); burst = bl.burst.checkReset(io); }
log.debug("sleep done", .{});
if (burst) { log.debug("Using Burst", .{}); try bl.burst.aquire(io); return; } } }
try bl.static.aquire(io); return; }};
pub const AuthType = enum { account, agent, none };
pub const Auth = struct { account: []const u8 = "", agent: []const u8 = "",};
pub const RequestOptions = struct { method: std.http.Method = .GET, auth: AuthType = .none, body: Body = .empty,
free_body_after_sending: bool = false,
pub const Body = union(enum) { empty: void, buffer: []u8, };
pub fn authorization(opts: *const RequestOptions, client: *const Client) HTTPClient.Request.Headers.Value { switch (opts.auth) { .account => return .{ .override = client.auth.account }, .agent => return .{ .override = client.auth.agent }, .none => return .{ .omit = {} }, } }};
pub const RequestError = error{ OutOfMemory, InvalidResponse, RateLimiterError,} || HTTPClient.RequestError || HTTPClient.Request.ReceiveHeadError || std.Uri.ParseError;
pub fn RawResponse(comptime T: type) type { return Io.Future(RequestError!json.Parsed(T));}
pub fn Response(comptime T: type) type { return RawResponse(models.Wrapper(T));}
pub const Client = struct { allocator: std.mem.Allocator, io: Io, limiter: BurstyLimiter,
base_url: []const u8, auth: Auth,
http: HTTPClient,
// Stats total_requests: std.atomic.Value(u64) = .init(0), successful_requests: std.atomic.Value(u64) = .init(0), total_latency: std.atomic.Value(u64) = .init(0), average_latency: std.atomic.Value(u64) = .init(0),
pub fn init( allocator: std.mem.Allocator, io: std.Io, opts: struct { base_url: []const u8 = "https://api.spacetraders.io/v2", auth: Auth = .{}, }, ) Client { return .{ .allocator = allocator, .io = io, .limiter = .{ .static = .init(.{}), .burst = .init(.{ .points = 30, .duration = 60_000 }), }, .base_url = opts.base_url, .auth = opts.auth, .http = .{ .allocator = allocator, .io = io }, }; }
pub fn deinit(client: *Client) void { client.http.deinit(); }
pub fn get( cl: *Client, comptime T: type, comptime path: []const u8, args: anytype, auth: AuthType, ) !RawResponse(T) { return cl.request(T, path, args, .{ .auth = auth }); }
pub fn getW( cl: *Client, comptime T: type, comptime path: []const u8, args: anytype, auth: AuthType, ) !Response(T) { return cl.get(models.Wrapper(T), path, args, auth); }
pub fn post( cl: *Client, comptime T: type, comptime path: []const u8, args: anytype, body: anytype, auth: AuthType, ) !RawResponse(T) { const buffer = try json.Stringify.valueAlloc(cl.allocator, body, .{}); log.debug("json body: {s}", .{buffer}); return cl.request(T, path, args, .{ .method = .POST, .auth = auth, .body = .{ .buffer = buffer }, .free_body_after_sending = true, }); }
pub fn postW( cl: *Client, comptime T: type, comptime path: []const u8, args: anytype, body: anytype, auth: AuthType, ) !Response(T) { return cl.post(models.Wrapper(T), path, args, body, auth); }
pub fn patch( cl: *Client, comptime T: type, comptime path: []const u8, args: anytype, body: anytype, auth: AuthType, ) !RawResponse(T) { const buffer = try json.Stringify.valueAlloc(cl.allocator, body, .{}); log.debug("json body: {s}", .{buffer}); return cl.request(T, path, args, .{ .method = .PATCH, .auth = auth, .body = .{ .buffer = buffer }, .free_body_after_sending = true, }); }
pub fn patchW( cl: *Client, comptime T: type, comptime path: []const u8, args: anytype, body: anytype, auth: AuthType, ) !Response(T) { return cl.patch(models.Wrapper(T), path, args, body, auth); }
pub fn request( client: *Client, comptime T: type, comptime path: []const u8, args: anytype, opts: RequestOptions, ) !RawResponse(T) { const path_fmt = try std.fmt.allocPrint(client.allocator, path, args); defer client.allocator.free(path_fmt);
const url = try std.fmt.allocPrint(client.allocator, "{s}{s}", .{ client.base_url, path_fmt });
const Wrapper = struct { fn call( cl: *Client, url_param: []const u8, opts_param: RequestOptions, ) RequestError!json.Parsed(T) { defer cl.allocator.free(url_param); cl.limiter.wait(cl.io) catch return error.RateLimiterError; return Client.requestRaw(cl, T, url_param, opts_param); } };
return client.io.concurrent( Wrapper.call, .{ client, url, opts }, ); }
const user_agent = blk: { const os = std.fmt.comptimePrint("{}", .{buildin.os.tag})[1..]; const arch = std.fmt.comptimePrint("{}", .{buildin.cpu.arch})[1..]; break :blk "space/0.1 (" ++ buildin.zig_version_string ++ "/" ++ os ++ "/" ++ arch ++ "; https://tangled.org/altagos.dev/space; +agent)"; };
pub fn requestRaw( client: *Client, comptime T: type, url: []const u8, opts: RequestOptions, ) RequestError!json.Parsed(T) { const uri = std.Uri.parse(url) catch |err| { log.err("Error parsing url: {} - url = {s}", .{ err, url }); return err; };
var req = try client.http.request(opts.method, uri, .{ .headers = .{ .authorization = opts.authorization(client), .user_agent = .{ .override = user_agent }, .content_type = .{ .override = "application/json" }, }, }); defer req.deinit();
log.debug("requesting: {s}", .{uri.path.percent_encoded});
const start = Io.Clock.now(.real, client.io);
switch (opts.body) { .empty => try req.sendBodiless(), .buffer => |body| { try req.sendBodyComplete(body); if (opts.free_body_after_sending) client.allocator.free(body); }, }
var redirect_buffer: [1024]u8 = undefined;
var response = try req.receiveHead(&redirect_buffer); const colour = blk: { if (@intFromEnum(response.head.status) >= 200 and @intFromEnum(response.head.status) < 300) { break :blk "\x1b[92m"; } else { break :blk "\x1b[1m\x1b[91m"; } };
_ = client.total_requests.fetchAdd(1, .seq_cst);
const latency: u64 = @intCast(start.durationTo(Io.Clock.now(.real, client.io)).toMilliseconds()); const old_average = client.average_latency.load(.seq_cst); var new_average: u64 = 0;
if (old_average == 0) { new_average = latency; } else { const total_reqs = client.total_requests.load(.seq_cst); new_average = old_average * (total_reqs - 1) / total_reqs + latency / total_reqs; }
client.average_latency.store(new_average, .seq_cst); _ = client.total_latency.fetchAdd(latency, .seq_cst);
log.debug( "\x1b[2m[latency = {d}ms path = {s}]\x1b[0m received {s}{d} {s}\x1b[0m", .{ latency, url[client.base_url.len..], colour, response.head.status, response.head.reason }, );
// var header_iter = response.head.iterateHeaders(); // while (header_iter.next()) |header| { // log.debug("{s}: {s}", .{ header.name, header.value }); // }
var decompress_buffer: [std.compress.flate.max_window_len]u8 = undefined; var transfer_buffer: [64]u8 = undefined; var decompress: std.http.Decompress = undefined;
const decompressed_body_reader = response.readerDecompressing( &transfer_buffer, &decompress, &decompress_buffer, );
const body = decompressed_body_reader.allocRemaining(client.allocator, .unlimited) catch return RequestError.OutOfMemory; defer client.allocator.free(body);
var reader: Io.Reader = .fixed(body); var json_reader: json.Reader = .init(client.allocator, &reader); defer json_reader.deinit();
const result = json.parseFromTokenSource(T, client.allocator, &json_reader, .{ .ignore_unknown_fields = true, }) catch |err| { log.err("Error parsing response: {} - Body:\n{s}", .{ err, body }); return RequestError.InvalidResponse; };
_ = client.successful_requests.fetchAdd(1, .seq_cst);
return result; }};