feat: add Ctx and worker to server
This commit is contained in:
dualfroz committed
1 parent
794a2465f5
commit
71086b2a09
1 file changed
+160
+160
@@ -7713,3 +7713,163 @@ test "server txn_op survives many real concurrent connections writing disjoint r
|
||||
try widgets.createIndex("owner");
|
||||
}
|
||||
|
||||
const address = try std.net.Address.parseIp("127.0.0.1", 0);
|
||||
var server = try Server.init(allocator, &db, address);
|
||||
try server.start();
|
||||
defer server.stop();
|
||||
|
||||
const client_mod = @import("client.zig");
|
||||
const threadCount = 6;
|
||||
const perThread = 40;
|
||||
|
||||
const Ctx = struct {
|
||||
allocator: Allocator,
|
||||
addr: std.net.Address,
|
||||
thread: i64,
|
||||
ok: bool = false,
|
||||
};
|
||||
|
||||
const worker = struct {
|
||||
fn run(ctx: *Ctx) void {
|
||||
var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return;
|
||||
defer client.close();
|
||||
var buf: [32]u8 = undefined;
|
||||
var i: i64 = 0;
|
||||
while (i < perThread) : (i += 1) {
|
||||
const ownerTag = std.fmt.bufPrint(&buf, "owner-{d}", .{ctx.thread}) catch return;
|
||||
const widgetPk: i64 = ctx.thread * 1_000_000 + i;
|
||||
const entries = [_]protocol.TxnOpEntry{
|
||||
.{ .verb = .put, .table = "widgets", .pk = .{ .int = widgetPk }, .fields = &.{.{ .name = "owner", .value = .{ .text = ownerTag } }} },
|
||||
.{ .verb = .put, .table = "tags", .pk = .{ .int = widgetPk }, .fields = &.{.{ .name = "seq", .value = .{ .int = i } }} },
|
||||
};
|
||||
const applied = client.txnOp(&entries) catch return;
|
||||
if (applied != 2) return;
|
||||
}
|
||||
ctx.ok = true;
|
||||
}
|
||||
}.run;
|
||||
|
||||
var contexts: [threadCount]Ctx = undefined;
|
||||
var threads: [threadCount]std.Thread = undefined;
|
||||
for (0..threadCount) |t| {
|
||||
contexts[t] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = @intCast(t) };
|
||||
threads[t] = try std.Thread.spawn(.{}, worker, .{&contexts[t]});
|
||||
}
|
||||
for (0..threadCount) |t| threads[t].join();
|
||||
for (0..threadCount) |t| try std.testing.expect(contexts[t].ok);
|
||||
|
||||
var widgets = try querymod.Table.open(&db, allocator, "widgets");
|
||||
defer widgets.deinit();
|
||||
var tags = try querymod.Table.open(&db, allocator, "tags");
|
||||
defer tags.deinit();
|
||||
|
||||
var total: usize = 0;
|
||||
for (0..threadCount) |t| {
|
||||
var buf: [32]u8 = undefined;
|
||||
const ownerTag = try std.fmt.bufPrint(&buf, "owner-{d}", .{t});
|
||||
const owned = try widgets.select(.{ .eq = .{ .field = "owner", .value = .{ .text = ownerTag } } });
|
||||
defer querymod.freeRows(allocator, owned);
|
||||
// Every widget this thread wrote must show up under its own index
|
||||
// entry, exactly `perThread` of them - not fewer (a lost write)
|
||||
// and not more (a duplicated one).
|
||||
try std.testing.expectEqual(@as(usize, perThread), owned.len);
|
||||
total += owned.len;
|
||||
}
|
||||
try std.testing.expectEqual(@as(usize, threadCount * perThread), total);
|
||||
|
||||
const allTags = try tags.select(null);
|
||||
defer querymod.freeRows(allocator, allTags);
|
||||
try std.testing.expectEqual(@as(usize, threadCount * perThread), allTags.len);
|
||||
}
|
||||
|
||||
test "server delivers a published message to a subscriber over real TCP" {
|
||||
const allocator = std.testing.allocator;
|
||||
var tmp = std.testing.tmpDir(.{});
|
||||
defer tmp.cleanup();
|
||||
const path = try tmp.dir.realpathAlloc(allocator, ".");
|
||||
defer allocator.free(path);
|
||||
|
||||
var db = try testDb(allocator, path);
|
||||
defer db.close();
|
||||
|
||||
const address = try std.net.Address.parseIp("127.0.0.1", 0);
|
||||
var server = try Server.init(allocator, &db, address);
|
||||
try server.start();
|
||||
defer server.stop();
|
||||
|
||||
const client_mod = @import("client.zig");
|
||||
var sub = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer sub.close();
|
||||
_ = try sub.subscribe(&.{"weather"});
|
||||
|
||||
var pub_ = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer pub_.close();
|
||||
const delivered = try pub_.publish("weather", "sunny");
|
||||
try std.testing.expectEqual(@as(u64, 1), delivered);
|
||||
|
||||
const msg = (try sub.nextMessage(allocator)).?;
|
||||
defer msg.deinit(allocator);
|
||||
try std.testing.expectEqualStrings("weather", msg.channel);
|
||||
try std.testing.expectEqualStrings("sunny", msg.payload);
|
||||
try std.testing.expectEqual(@as(u64, 1), server.messagesPublishedTotal.get());
|
||||
}
|
||||
|
||||
test "server delivers to a subscriber's own copy per channel it named, every subscriber on a channel reached" {
|
||||
const allocator = std.testing.allocator;
|
||||
var tmp = std.testing.tmpDir(.{});
|
||||
defer tmp.cleanup();
|
||||
const path = try tmp.dir.realpathAlloc(allocator, ".");
|
||||
defer allocator.free(path);
|
||||
|
||||
var db = try testDb(allocator, path);
|
||||
defer db.close();
|
||||
|
||||
const address = try std.net.Address.parseIp("127.0.0.1", 0);
|
||||
var server = try Server.init(allocator, &db, address);
|
||||
try server.start();
|
||||
defer server.stop();
|
||||
|
||||
const client_mod = @import("client.zig");
|
||||
var subA = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer subA.close();
|
||||
_ = try subA.subscribe(&.{ "orders", "alerts" });
|
||||
|
||||
var subB = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer subB.close();
|
||||
_ = try subB.subscribe(&.{"orders"});
|
||||
|
||||
var pub_ = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer pub_.close();
|
||||
try std.testing.expectEqual(@as(u64, 2), try pub_.publish("orders", "order-1"));
|
||||
try std.testing.expectEqual(@as(u64, 1), try pub_.publish("alerts", "disk-full"));
|
||||
|
||||
const aFirst = (try subA.nextMessage(allocator)).?;
|
||||
defer aFirst.deinit(allocator);
|
||||
try std.testing.expectEqualStrings("orders", aFirst.channel);
|
||||
try std.testing.expectEqualStrings("order-1", aFirst.payload);
|
||||
const aSecond = (try subA.nextMessage(allocator)).?;
|
||||
defer aSecond.deinit(allocator);
|
||||
try std.testing.expectEqualStrings("alerts", aSecond.channel);
|
||||
try std.testing.expectEqualStrings("disk-full", aSecond.payload);
|
||||
|
||||
const bOnly = (try subB.nextMessage(allocator)).?;
|
||||
defer bOnly.deinit(allocator);
|
||||
try std.testing.expectEqualStrings("orders", bOnly.channel);
|
||||
try std.testing.expectEqualStrings("order-1", bOnly.payload);
|
||||
}
|
||||
|
||||
test "server publish to a channel with no subscribers returns 0 without crashing, over real TCP" {
|
||||
const allocator = std.testing.allocator;
|
||||
var tmp = std.testing.tmpDir(.{});
|
||||
defer tmp.cleanup();
|
||||
const path = try tmp.dir.realpathAlloc(allocator, ".");
|
||||
defer allocator.free(path);
|
||||
|
||||
var db = try testDb(allocator, path);
|
||||
defer db.close();
|
||||
|
||||
const address = try std.net.Address.parseIp("127.0.0.1", 0);
|
||||
var server = try Server.init(allocator, &db, address);
|
||||
try server.start();
|
||||
defer server.stop();
|
||||
|
||||
Reference in new issue
Block a user