feat: extend main with runRemoteUnsubscribe and runRemoteWindow
This commit is contained in:
dualfroz committed
1 parent
29b2750464
commit
422f7c1639
1 file changed
+173
+173
@@ -2906,6 +2906,179 @@ fn runRemoteSubscribe(allocator: std.mem.Allocator, stdout: anytype, stderr: any
|
||||
/// `<subscriber-id>` its `remote-subscribe` printed - the client-side
|
||||
/// counterpart of `server.zig`'s `unsubscribe` opcode. Unlike
|
||||
/// `remote-subscribe`, this connection is an ordinary request/response
|
||||
/// round trip: it prints how many channels were actually removed and
|
||||
/// exits, never blocking. `<subscriber-id>` must parse as a `u64` - a
|
||||
/// malformed one is reported the same way a malformed `--limit` is
|
||||
/// elsewhere in this file, before ever reaching for the network.
|
||||
fn runRemoteUnsubscribe(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, addressSpec: []const u8, rest: []const []const u8) !void {
|
||||
if (rest.len < 2) return fail(stderr, "remote-unsubscribe needs <subscriber-id> <channel>[,channel...] [--user NAME --pass PASS] [--token TOKEN]");
|
||||
const address = try parseHostPort(stdout, addressSpec);
|
||||
|
||||
const subscriberId = std.fmt.parseInt(u64, rest[0], 10) catch return fail(stderr, "invalid <subscriber-id> (expected a non-negative integer)");
|
||||
|
||||
var client = cobalt.Client.connect(allocator, address) catch |err| {
|
||||
try stderr.print("error: could not connect to {s}: {s}\n", .{ addressSpec, @errorName(err) });
|
||||
std.process.exit(1);
|
||||
};
|
||||
defer client.close();
|
||||
|
||||
const channels = parseChannelList(allocator, rest[1]) catch return fail(stderr, "invalid channel list (empty channel name)");
|
||||
defer allocator.free(channels);
|
||||
|
||||
const authFlags = extractAuthFlags(allocator, rest[2..]) catch return fail(stderr, "invalid --user/--pass/--token value");
|
||||
defer allocator.free(authFlags.rest);
|
||||
|
||||
try authenticateIfRequested(&client, stderr, authFlags.auth);
|
||||
|
||||
const removed = client.unsubscribe(subscriberId, channels) catch |err| {
|
||||
try stderr.print("error: {s}\n", .{@errorName(err)});
|
||||
std.process.exit(1);
|
||||
};
|
||||
try stdout.print("{d}\n", .{removed});
|
||||
}
|
||||
|
||||
fn runRemoteWindow(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, addressSpec: []const u8, rest: []const []const u8) !void {
|
||||
if (rest.len < 2) return fail(stderr, "remote-window needs <table> <kind[:field]>[,<kind[:field]>...] [expr...] [partition <field>[,field...]] [order <field[:desc]>[,field...]] [--limit N] [--offset N] [--range]");
|
||||
const address = try parseHostPort(stdout, addressSpec);
|
||||
|
||||
var client = cobalt.Client.connect(allocator, address) catch |err| {
|
||||
try stderr.print("error: could not connect to {s}: {s}\n", .{ addressSpec, @errorName(err) });
|
||||
std.process.exit(1);
|
||||
};
|
||||
defer client.close();
|
||||
|
||||
const table = rest[0];
|
||||
const specs = parseWindowSpecs(allocator, rest[1]) catch return fail(stderr, "invalid window spec (expected row_number | rank | dense_rank | sum:field | avg:field | count | count:field | count_distinct:field | min:field | max:field | first_value:field | last_value:field | nth_value:field:n[:default] | lag:field[:offset[:default]] | lead:field[:offset[:default]], comma-separated)");
|
||||
defer allocator.free(specs);
|
||||
|
||||
const authFlags = extractAuthFlags(allocator, rest[2..]) catch return fail(stderr, "invalid --user/--pass/--token value");
|
||||
defer allocator.free(authFlags.rest);
|
||||
|
||||
try authenticateIfRequested(&client, stderr, authFlags.auth);
|
||||
|
||||
const flags = extractPageFlags(allocator, authFlags.rest) catch return fail(stderr, "invalid --limit/--offset value");
|
||||
defer allocator.free(flags.rest);
|
||||
|
||||
const rangeFlag = try extractRangeFlag(allocator, flags.rest);
|
||||
defer allocator.free(rangeFlag.rest);
|
||||
const frame: cobalt.window.FrameMode = if (rangeFlag.range) .range else .rows;
|
||||
|
||||
const split = splitWindowArgs(rangeFlag.rest);
|
||||
|
||||
const exprText = try std.mem.join(allocator, " ", split.whereArgs);
|
||||
defer allocator.free(exprText);
|
||||
|
||||
var partitionBy: [][]const u8 = &.{};
|
||||
defer allocator.free(partitionBy);
|
||||
if (split.partitionArgs) |partitionArgs| {
|
||||
if (partitionArgs.len != 1) return fail(stderr, "partition requires exactly one comma-separated field list");
|
||||
partitionBy = parsePartitionList(allocator, partitionArgs[0]) catch return fail(stderr, "invalid partition field list");
|
||||
}
|
||||
|
||||
var orderBy: []cobalt.query.OrderBy = &.{};
|
||||
defer allocator.free(orderBy);
|
||||
if (split.orderArgs) |orderArgs| {
|
||||
if (orderArgs.len != 1) return fail(stderr, "order requires exactly one comma-separated field list");
|
||||
orderBy = parseOrderByList(allocator, orderArgs[0]) catch return fail(stderr, "invalid order field list");
|
||||
}
|
||||
|
||||
const results = client.tableWindow(allocator, table, specs, partitionBy, orderBy, flags.page.offset, flags.page.limit, exprText, frame) catch |err| {
|
||||
try stderr.print("error: {s}\n", .{@errorName(err)});
|
||||
std.process.exit(1);
|
||||
};
|
||||
defer cobalt.protocol.freeWindowRows(allocator, results);
|
||||
|
||||
for (results) |wr| {
|
||||
for (wr.row.fields.items, 0..) |f, j| {
|
||||
if (j > 0) try stdout.writeAll("\t");
|
||||
try stdout.print("{s}=", .{f.name});
|
||||
try printValue(stdout, f.value);
|
||||
}
|
||||
if (wr.row.fields.items.len > 0) try stdout.writeAll("\t");
|
||||
for (specs, 0..) |spec, specIdx| {
|
||||
if (specIdx > 0) try stdout.writeAll("\t");
|
||||
try printWindowLabel(stdout, spec);
|
||||
try stdout.writeAll("=");
|
||||
try printValue(stdout, wr.values[specIdx]);
|
||||
}
|
||||
try stdout.writeAll("\n");
|
||||
}
|
||||
}
|
||||
|
||||
/// The network counterpart of `runTxn`: parses the exact same
|
||||
/// `put|del <table> <pk> [field=value ...]` script format as the local
|
||||
/// `txn <script-file>` command, but builds a list of
|
||||
/// `protocol.TxnOpEntry` and commits it in one round trip through
|
||||
/// `client.zig`'s `txnOp` (the `txn_op` opcode) instead of staging against
|
||||
/// a local `TableTxn` directly - so a script that used to require a
|
||||
/// process holding the `Db` can now run against a remote `cobalt serve`.
|
||||
fn runRemoteTxn(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, addressSpec: []const u8, rest: []const []const u8) !void {
|
||||
if (rest.len < 1) return fail(stderr, "remote-txn needs <script-file> [--user NAME --pass PASS] [--token TOKEN]");
|
||||
const address = try parseHostPort(stdout, addressSpec);
|
||||
|
||||
const authFlags = extractAuthFlags(allocator, rest[1..]) catch return fail(stderr, "invalid --user/--pass/--token value");
|
||||
defer allocator.free(authFlags.rest);
|
||||
|
||||
const contents = std.fs.cwd().readFileAlloc(allocator, rest[0], 16 * 1024 * 1024) catch |err| {
|
||||
try stderr.print("failed to read script '{s}': {s}\n", .{ rest[0], @errorName(err) });
|
||||
std.process.exit(1);
|
||||
};
|
||||
defer allocator.free(contents);
|
||||
|
||||
var entries = std.ArrayList(cobalt.protocol.TxnOpEntry).init(allocator);
|
||||
defer {
|
||||
for (entries.items) |e| allocator.free(e.fields);
|
||||
entries.deinit();
|
||||
}
|
||||
|
||||
var lineNo: usize = 0;
|
||||
var lines = std.mem.splitScalar(u8, contents, '\n');
|
||||
while (lines.next()) |rawLine| {
|
||||
lineNo += 1;
|
||||
const line = std.mem.trim(u8, rawLine, " \t\r");
|
||||
if (line.len == 0 or line[0] == '#') continue;
|
||||
|
||||
var tok = std.mem.tokenizeScalar(u8, line, ' ');
|
||||
const verb = tok.next() orelse continue;
|
||||
const tableName = tok.next() orelse {
|
||||
try stderr.print("line {d}: missing table name\n", .{lineNo});
|
||||
std.process.exit(2);
|
||||
};
|
||||
const pkText = tok.next() orelse {
|
||||
try stderr.print("line {d}: missing primary key\n", .{lineNo});
|
||||
std.process.exit(2);
|
||||
};
|
||||
const pk = parseValue(pkText) catch {
|
||||
try stderr.print("line {d}: invalid primary key '{s}'\n", .{ lineNo, pkText });
|
||||
std.process.exit(2);
|
||||
};
|
||||
|
||||
if (std.mem.eql(u8, verb, "put")) {
|
||||
var fields = std.ArrayList(Field).init(allocator);
|
||||
errdefer fields.deinit();
|
||||
while (tok.next()) |raw| {
|
||||
const f = parseField(raw) catch {
|
||||
try stderr.print("line {d}: invalid field '{s}'\n", .{ lineNo, raw });
|
||||
std.process.exit(2);
|
||||
};
|
||||
try fields.append(f);
|
||||
}
|
||||
try entries.append(.{ .verb = .put, .table = tableName, .pk = pk, .fields = try fields.toOwnedSlice() });
|
||||
} else if (std.mem.eql(u8, verb, "del")) {
|
||||
try entries.append(.{ .verb = .del, .table = tableName, .pk = pk, .fields = &.{} });
|
||||
} else {
|
||||
try stderr.print("line {d}: unknown verb '{s}' (expected put/del)\n", .{ lineNo, verb });
|
||||
std.process.exit(2);
|
||||
}
|
||||
}
|
||||
|
||||
if (entries.items.len == 0) return fail(stderr, "txn script had no operations");
|
||||
|
||||
var client = cobalt.Client.connect(allocator, address) catch |err| {
|
||||
try stderr.print("error: could not connect to {s}: {s}\n", .{ addressSpec, @errorName(err) });
|
||||
std.process.exit(1);
|
||||
};
|
||||
defer client.close();
|
||||
}
|
||||
|
||||
fn runVerify(allocator: std.mem.Allocator, stdout: anytype, dir: []const u8, count: usize) !void {
|
||||
|
||||
Reference in new issue
Block a user