feat: add RemoteExport and deinit
This commit is contained in:
1 parent
a5ba9ed3db
commit
cc546fe216
1 file changed
+158
+158
@@ -0,0 +1,158 @@
|
|||||||
|
const std = @import("std");
|
||||||
|
const protocol = @import("protocol.zig");
|
||||||
|
const dbmod = @import("db.zig");
|
||||||
|
const rowmod = @import("row.zig");
|
||||||
|
const authmod = @import("auth.zig");
|
||||||
|
const ttlmod = @import("ttl.zig");
|
||||||
|
const collmod = @import("collections.zig");
|
||||||
|
const aggregatemod = @import("aggregate.zig");
|
||||||
|
const windowmod = @import("window.zig");
|
||||||
|
const querymod = @import("query.zig");
|
||||||
|
|
||||||
|
const Allocator = std.mem.Allocator;
|
||||||
|
|
||||||
|
pub const ClientError = error{
|
||||||
|
ServerBadRequest,
|
||||||
|
ServerInternal,
|
||||||
|
ServerConflict,
|
||||||
|
ServerRateLimited,
|
||||||
|
ServerUnauthorized,
|
||||||
|
ServerForbidden,
|
||||||
|
ServerUserNotFound,
|
||||||
|
ServerKeyNotFound,
|
||||||
|
ServerCollectionNotFound,
|
||||||
|
UnexpectedStatus,
|
||||||
|
};
|
||||||
|
|
||||||
|
/// Result of a remote `table export`: the row count `dump.zig`'s exporter
|
||||||
|
/// reported plus the rendered bytes, independently allocated so they
|
||||||
|
/// outlive the response frame they arrived in.
|
||||||
|
pub const RemoteExport = struct {
|
||||||
|
rowCount: u64,
|
||||||
|
content: []u8,
|
||||||
|
|
||||||
|
pub fn deinit(self: RemoteExport, allocator: Allocator) void {
|
||||||
|
allocator.free(self.content);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
fn statusToError(status: protocol.Status) ClientError {
|
||||||
|
return switch (status) {
|
||||||
|
.bad_request => error.ServerBadRequest,
|
||||||
|
.internal => error.ServerInternal,
|
||||||
|
.conflict => error.ServerConflict,
|
||||||
|
.rate_limited => error.ServerRateLimited,
|
||||||
|
.unauthorized => error.ServerUnauthorized,
|
||||||
|
.forbidden => error.ServerForbidden,
|
||||||
|
else => error.UnexpectedStatus,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A blocking client for the wire protocol in `protocol.zig`. Each call
|
||||||
|
/// writes one request frame and reads exactly one response frame, so a
|
||||||
|
/// `Client` must not be shared across threads without external locking.
|
||||||
|
pub const Client = struct {
|
||||||
|
allocator: Allocator,
|
||||||
|
stream: std.net.Stream,
|
||||||
|
/// The session token from the last successful `authenticate` call, if
|
||||||
|
/// the server had `enableSessions` on (see server.zig) - null on a
|
||||||
|
/// server without sessions, or before any successful `authenticate`.
|
||||||
|
/// Owned by the `Client`, freed on `close()` and on every subsequent
|
||||||
|
/// `authenticate`/`authenticateWithToken` call.
|
||||||
|
token: ?[]u8 = null,
|
||||||
|
|
||||||
|
pub fn connect(allocator: Allocator, address: std.net.Address) !Client {
|
||||||
|
const stream = try std.net.tcpConnectToAddress(address);
|
||||||
|
return .{ .allocator = allocator, .stream = stream };
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn close(self: *Client) void {
|
||||||
|
if (self.token) |t| self.allocator.free(t);
|
||||||
|
self.stream.close();
|
||||||
|
}
|
||||||
|
|
||||||
|
fn setToken(self: *Client, payload: []const u8) !void {
|
||||||
|
if (self.token) |old| self.allocator.free(old);
|
||||||
|
self.token = if (payload.len > 0) try self.allocator.dupe(u8, payload) else null;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The session token captured by the last successful `authenticate`,
|
||||||
|
/// if any (see `token` above). Callers that want to reconnect later
|
||||||
|
/// without a password should dupe this before the `Client` closes.
|
||||||
|
pub fn sessionToken(self: *Client) ?[]const u8 {
|
||||||
|
return self.token;
|
||||||
|
}
|
||||||
|
|
||||||
|
fn roundTrip(self: *Client, op: protocol.Op, payload: []const u8) !protocol.Frame {
|
||||||
|
try protocol.writeFrame(self.stream.writer(), @intFromEnum(op), payload);
|
||||||
|
return protocol.readFrame(self.allocator, self.stream.reader());
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Authenticates this connection against a server that has auth
|
||||||
|
/// enabled (see `server.zig`'s `enableAuth`). Every request on this
|
||||||
|
/// stream after a successful call is dispatched under the role
|
||||||
|
/// returned by the server's `auth.Store.verify`; a server without auth
|
||||||
|
/// enabled always answers `ok` regardless of the credentials given. If
|
||||||
|
/// the server also has sessions enabled (`enableSessions`), the
|
||||||
|
/// response carries a fresh token captured into `sessionToken()` -
|
||||||
|
/// otherwise `sessionToken()` reads null afterwards, same as before
|
||||||
|
/// this call was ever made.
|
||||||
|
pub fn authenticate(self: *Client, username: []const u8, password: []const u8) !void {
|
||||||
|
const payload = try protocol.encodeAuth(self.allocator, username, password);
|
||||||
|
defer self.allocator.free(payload);
|
||||||
|
var frame = try self.roundTrip(.auth, payload);
|
||||||
|
defer frame.deinit();
|
||||||
|
const status = try protocol.Status.fromByte(frame.tag);
|
||||||
|
if (status != .ok) return statusToError(status);
|
||||||
|
try self.setToken(frame.payload);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Authenticates this connection using a token obtained from another
|
||||||
|
/// connection's `sessionToken()`, instead of a username/password pair
|
||||||
|
/// (see server.zig's `auth_token` opcode) - skips the password
|
||||||
|
/// entirely and resumes the same role until the session expires. A
|
||||||
|
/// server without sessions enabled accepts any token unconditionally,
|
||||||
|
/// matching `authenticate`'s behavior against a server without auth.
|
||||||
|
pub fn authenticateWithToken(self: *Client, token: []const u8) !void {
|
||||||
|
const payload = try protocol.encodeAuthToken(self.allocator, token);
|
||||||
|
defer self.allocator.free(payload);
|
||||||
|
var frame = try self.roundTrip(.auth_token, payload);
|
||||||
|
defer frame.deinit();
|
||||||
|
const status = try protocol.Status.fromByte(frame.tag);
|
||||||
|
if (status != .ok) return statusToError(status);
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn ping(self: *Client) !void {
|
||||||
|
var frame = try self.roundTrip(.ping, "");
|
||||||
|
defer frame.deinit();
|
||||||
|
const status = try protocol.Status.fromByte(frame.tag);
|
||||||
|
if (status != .ok) return statusToError(status);
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn get(self: *Client, allocator: Allocator, key: []const u8) !?[]u8 {
|
||||||
|
var frame = try self.roundTrip(.get, key);
|
||||||
|
defer frame.deinit();
|
||||||
|
const status = try protocol.Status.fromByte(frame.tag);
|
||||||
|
return switch (status) {
|
||||||
|
.ok => try allocator.dupe(u8, frame.payload),
|
||||||
|
.not_found => null,
|
||||||
|
else => statusToError(status),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn set(self: *Client, key: []const u8, value: []const u8) !void {
|
||||||
|
const payload = try protocol.encodeSet(self.allocator, key, value);
|
||||||
|
defer self.allocator.free(payload);
|
||||||
|
var frame = try self.roundTrip(.set, payload);
|
||||||
|
defer frame.deinit();
|
||||||
|
const status = try protocol.Status.fromByte(frame.tag);
|
||||||
|
if (status != .ok) return statusToError(status);
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn delete(self: *Client, key: []const u8) !void {
|
||||||
|
var frame = try self.roundTrip(.del, key);
|
||||||
|
defer frame.deinit();
|
||||||
|
const status = try protocol.Status.fromByte(frame.tag);
|
||||||
|
if (status != .ok) return statusToError(status);
|
||||||
|
}
|
||||||
|
|
||||||
Reference in new issue
Block a user