summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/main.zig12
-rw-r--r--src/root.zig5
-rw-r--r--src/server/Client.zig (renamed from src/server/client.zig)0
-rw-r--r--src/server/Server.zig305
-rw-r--r--src/server/main.zig329
-rw-r--r--src/server/test.zig3
6 files changed, 336 insertions, 318 deletions
diff --git a/src/main.zig b/src/main.zig
index 7e4974c..6b2de8b 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -23,7 +23,7 @@ pub fn main() !void {
'a',
std.fmt.comptimePrint(
"Address to bind to (default: {s})",
- .{std.meta.fieldInfo(zits.Server.ServerInfo, .host).defaultValue().?},
+ .{std.meta.fieldInfo(zits.Server.Message.ServerInfo, .host).defaultValue().?},
),
),
yazap.Arg.singleValueOption(
@@ -31,7 +31,7 @@ pub fn main() !void {
'p',
std.fmt.comptimePrint(
"Port to listen on (default: {d})",
- .{std.meta.fieldInfo(zits.Server.ServerInfo, .port).defaultValue().?},
+ .{std.meta.fieldInfo(zits.Server.Message.ServerInfo, .port).defaultValue().?},
),
),
yazap.Arg.singleValueOption(
@@ -52,9 +52,9 @@ pub fn main() !void {
const matches = try app.parseProcess(io);
if (matches.subcommandMatches("serve")) |serve_matches| {
- var info: zits.Server.ServerInfo = .{
- .server_id = zits.Server.createId(),
- .server_name = zits.Server.createName(),
+ var info: zits.Server.Message.ServerInfo = .{
+ .server_id = zits.Server.default_id,
+ .server_name = zits.Server.default_name,
.version = "zits-master",
.max_payload = 1048576,
.headers = true,
@@ -67,7 +67,7 @@ pub fn main() !void {
info.server_name = name;
}
- try Server.main(gpa, info);
+ try @import("./server/main.zig").main(gpa, info);
return;
}
}
diff --git a/src/root.zig b/src/root.zig
index e46af60..49631cb 100644
--- a/src/root.zig
+++ b/src/root.zig
@@ -1,2 +1,3 @@
-pub const MessageParser = @import("server/message_parser.zig");
-pub const Server = @import("server/main.zig");
+const MessageParser = @import("server/message_parser.zig");
+
+pub const Server = @import("server/Server.zig");
diff --git a/src/server/client.zig b/src/server/Client.zig
index 2ce3c38..2ce3c38 100644
--- a/src/server/client.zig
+++ b/src/server/Client.zig
diff --git a/src/server/Server.zig b/src/server/Server.zig
new file mode 100644
index 0000000..ebbac19
--- /dev/null
+++ b/src/server/Server.zig
@@ -0,0 +1,305 @@
+const std = @import("std");
+const Allocator = std.mem.Allocator;
+const ArrayList = std.ArrayList;
+const AutoHashMapUnmanaged = std.AutoHashMapUnmanaged;
+
+const Io = std.Io;
+const Group = Io.Group;
+const IpAddress = std.Io.net.IpAddress;
+const Mutex = Io.Mutex;
+const Queue = Io.Queue;
+const Stream = std.Io.net.Stream;
+
+const message_parser = @import("./message_parser.zig");
+pub const MessageType = message_parser.MessageType;
+pub const Message = message_parser.Message;
+const ServerInfo = Message.ServerInfo;
+pub const Client = @import("./Client.zig");
+const Server = @This();
+
+pub const Subscription = struct {
+ subject: []const u8,
+ client_id: usize,
+ sid: []const u8,
+
+ fn deinit(self: Subscription, alloc: Allocator) void {
+ alloc.free(self.subject);
+ alloc.free(self.sid);
+ }
+};
+
+const eql = std.mem.eql;
+const log = std.log;
+const panic = std.debug.panic;
+
+info: ServerInfo,
+clients: AutoHashMapUnmanaged(usize, *Client) = .empty,
+
+subs_lock: Mutex = .init,
+subscriptions: ArrayList(Subscription) = .empty,
+
+pub fn deinit(server: *Server, io: Io, alloc: Allocator) void {
+ server.subs_lock.lockUncancelable(io);
+ defer server.subs_lock.unlock(io);
+ for (server.subscriptions.items) |sub| {
+ sub.deinit(alloc);
+ }
+ server.subscriptions.deinit(alloc);
+ server.clients.deinit(alloc);
+}
+
+pub fn start(server: *Server, io: Io, gpa: Allocator) !void {
+ var tcp_server = try IpAddress.listen(try IpAddress.parse(
+ server.info.host,
+ server.info.port,
+ ), io, .{ .reuse_address = true });
+ defer tcp_server.deinit(io);
+ log.debug("Server headers: {s}", .{if (server.info.headers) "true" else "false"});
+ log.debug("Server max payload: {d}", .{server.info.max_payload});
+ log.info("Server ID: {s}", .{server.info.server_id});
+ log.info("Server name: {s}", .{server.info.server_name});
+ log.info("Server listening on {s}:{d}", .{ server.info.host, server.info.port });
+
+ var client_group: Group = .init;
+ defer client_group.cancel(io);
+
+ var id: usize = 0;
+ while (true) : (id +%= 1) {
+ if (server.clients.contains(id)) continue;
+ log.debug("Accepting next client", .{});
+ const stream = try tcp_server.accept(io);
+ log.debug("Accepted connection {d}", .{id});
+ _ = client_group.concurrent(io, handleConnectionInfallible, .{ server, gpa, io, id, stream }) catch {
+ log.err("Could not start concurrent handler for {d}", .{id});
+ stream.close(io);
+ };
+ }
+}
+
+fn addClient(server: *Server, allocator: Allocator, id: usize, client: *Client) !void {
+ try server.clients.put(allocator, id, client);
+}
+
+fn removeClient(server: *Server, io: Io, allocator: Allocator, id: usize) void {
+ server.subs_lock.lockUncancelable(io);
+ defer server.subs_lock.unlock(io);
+ if (server.clients.remove(id)) {
+ const len = server.subscriptions.items.len;
+ for (0..len) |from_end| {
+ const i = len - from_end - 1;
+ const sub = server.subscriptions.items[i];
+ if (sub.client_id == id) {
+ sub.deinit(allocator);
+ _ = server.subscriptions.swapRemove(i);
+ }
+ }
+ }
+}
+
+fn handleConnectionInfallible(server: *Server, server_allocator: Allocator, io: Io, id: usize, stream: Stream) void {
+ handleConnection(server, server_allocator, io, id, stream) catch |err| {
+ log.err("Failed processing client {d}: {any}", .{ id, err });
+ };
+}
+
+fn handleConnection(server: *Server, server_allocator: Allocator, io: Io, id: usize, stream: Stream) !void {
+ defer stream.close(io);
+
+ // TODO: use a client allocator for things that should only live for as long as the client?
+ // I had this before, but it seemed to have made lifetimes harder to track.
+ // Messages made sense to parse using a client allocator, but that makes it hard to free
+ // messages when done processing them (usually outside the client process, ie: publish).
+
+ // Set up client writer
+ // TODO: how many bytes can fit in a network write syscall? cat /proc/sys/net/core/wmem_max
+ var w_buffer: [1024 * 16]u8 = undefined;
+ var writer = stream.writer(io, &w_buffer);
+ const out = &writer.interface;
+
+ // Set up client reader
+ // TODO: how many bytes can fit in a network read syscall? cat /proc/sys/net/core/rmem_max
+ var r_buffer: [1024 * 16]u8 = undefined;
+ var reader = stream.reader(io, &r_buffer);
+ const in = &reader.interface;
+
+ // Set up buffer queue
+ var qbuf: [8]Message = undefined;
+ var queue: Queue(Message) = .init(&qbuf);
+ defer {
+ queue.close(io);
+ while (queue.getOne(io)) |msg| {
+ switch (msg) {
+ .msg => |m| m.deinit(server_allocator),
+ .hmsg => |h| h.deinit(server_allocator),
+ else => {},
+ }
+ } else |_| {}
+ }
+
+ // Create client
+ var client: Client = .init(null, &queue, in, out);
+ defer client.deinit(server_allocator);
+
+ try server.addClient(server_allocator, id, &client);
+ defer server.removeClient(io, server_allocator, id);
+
+ // Do initial handshake with client
+ try queue.putOne(io, .{ .info = server.info });
+
+ var client_task = try io.concurrent(Client.start, .{ &client, io, server_allocator });
+ defer client_task.cancel(io) catch {};
+
+ // Messages are owned by the server after they are received from the client
+ while (client.next(server_allocator)) |msg| {
+ switch (msg) {
+ .ping => {
+ // Respond to ping with pong.
+ try client.send(io, .pong);
+ },
+ .@"pub", .hpub => {
+ defer switch (msg) {
+ .@"pub" => |pb| pb.deinit(server_allocator),
+ .hpub => |hp| hp.deinit(server_allocator),
+ else => unreachable,
+ };
+ try server.publishMessage(io, server_allocator, &client, msg);
+ },
+ .sub => |sub| {
+ defer sub.deinit(server_allocator);
+ try server.subscribe(io, server_allocator, id, sub);
+ },
+ .unsub => |unsub| {
+ defer unsub.deinit(server_allocator);
+ try server.unsubscribe(io, server_allocator, id, unsub);
+ },
+ .connect => |connect| {
+ if (client.connect) |*current| {
+ current.deinit(server_allocator);
+ }
+ client.connect = connect;
+ },
+ else => |e| {
+ panic("Unimplemented message: {any}\n", .{e});
+ },
+ }
+ } else |err| switch (err) {
+ error.EndOfStream => {
+ log.debug("Client {d} disconnected", .{id});
+ },
+ else => {
+ return err;
+ },
+ }
+}
+
+fn subjectMatches(sub_subject: []const u8, pub_subject: []const u8) bool {
+ // TODO: assert that sub_subject and pub_subject are valid.
+ var sub_iter = std.mem.splitScalar(u8, sub_subject, '.');
+ var pub_iter = std.mem.splitScalar(u8, pub_subject, '.');
+
+ while (sub_iter.next()) |st| {
+ const pt = pub_iter.next() orelse return false;
+
+ if (eql(u8, st, ">")) return true;
+
+ if (!eql(u8, st, "*") and !eql(u8, st, pt)) {
+ return false;
+ }
+ }
+
+ return pub_iter.next() == null;
+}
+
+test subjectMatches {
+ const expect = std.testing.expect;
+ try expect(subjectMatches("foo", "foo"));
+ try expect(!subjectMatches("foo", "bar"));
+
+ try expect(subjectMatches("foo.*", "foo.bar"));
+ try expect(!subjectMatches("foo.*", "foo"));
+ try expect(!subjectMatches("foo.>", "foo"));
+
+ // the wildcard subscriptions foo.*.quux and foo.> both match foo.bar.quux, but only the latter matches foo.bar.baz.
+ try expect(subjectMatches("foo.*.quux", "foo.bar.quux"));
+ try expect(subjectMatches("foo.>", "foo.bar.quux"));
+ try expect(!subjectMatches("foo.*.quux", "foo.bar.baz"));
+ try expect(subjectMatches("foo.>", "foo.bar.baz"));
+}
+
+fn publishMessage(server: *Server, io: Io, alloc: Allocator, source_client: *Client, msg: Message) !void {
+ errdefer {
+ if (source_client.connect) |c| {
+ if (c.verbose) {
+ source_client.send(io, .{ .@"-err" = "Slow Consumer" }) catch {};
+ }
+ }
+ }
+ const subject = switch (msg) {
+ .@"pub" => |pb| pb.subject,
+ .hpub => |hp| hp.@"pub".subject,
+ else => unreachable,
+ };
+ try server.subs_lock.lock(io);
+ defer server.subs_lock.unlock(io);
+ for (server.subscriptions.items) |subscription| {
+ if (subjectMatches(subscription.subject, subject)) {
+ const client = server.clients.get(subscription.client_id) orelse {
+ log.debug("Trying to publish to a client that no longer exists: {d}\n", .{subscription.client_id});
+ continue;
+ };
+
+ switch (msg) {
+ .@"pub" => |pb| client.send(io, .{
+ .msg = try pb.toMsg(alloc, subscription.sid),
+ }) catch |err| switch (err) {
+ error.Canceled => return err,
+ else => {},
+ },
+ .hpub => |hp| client.send(io, .{ .hmsg = try hp.toHMsg(
+ alloc,
+ subscription.sid,
+ ) }) catch |err| switch (err) {
+ error.Canceled => return err,
+ else => {},
+ },
+ else => unreachable,
+ }
+ }
+ }
+ if (source_client.connect) |c| {
+ if (c.verbose) {
+ source_client.send(io, .@"+ok") catch {};
+ }
+ }
+}
+
+fn subscribe(server: *Server, io: Io, gpa: Allocator, id: usize, msg: Message.Sub) !void {
+ try server.subs_lock.lock(io);
+ defer server.subs_lock.unlock(io);
+ const subject = try gpa.dupe(u8, msg.subject);
+ errdefer gpa.free(subject);
+ const sid = try gpa.dupe(u8, msg.sid);
+ errdefer gpa.free(sid);
+ try server.subscriptions.append(gpa, .{
+ .subject = subject,
+ .client_id = id,
+ .sid = sid,
+ });
+}
+
+fn unsubscribe(server: *Server, io: Io, gpa: Allocator, id: usize, msg: Message.Unsub) !void {
+ try server.subs_lock.lock(io);
+ defer server.subs_lock.unlock(io);
+ const len = server.subscriptions.items.len;
+ for (0..len) |from_end| {
+ const i = len - from_end - 1;
+ const sub = server.subscriptions.items[i];
+ if (sub.client_id == id and eql(u8, sub.sid, msg.sid)) {
+ sub.deinit(gpa);
+ _ = server.subscriptions.swapRemove(i);
+ }
+ }
+}
+
+pub const default_id = "server-id-123";
+pub const default_name = "Zits Server";
diff --git a/src/server/main.zig b/src/server/main.zig
index 4e47c10..1aaf572 100644
--- a/src/server/main.zig
+++ b/src/server/main.zig
@@ -1,48 +1,32 @@
const std = @import("std");
-const builtin = @import("builtin");
-const Message = @import("./message_parser.zig").Message;
-pub const ServerInfo = Message.ServerInfo;
+const Allocator = std.mem.Allocator;
+const AtomicValue = std.atomic.Value;
+const DebugAllocator = std.heap.DebugAllocator;
+const Sigaction = std.posix.Sigaction;
-const Client = @import("./client.zig");
-const Server = @This();
+const Io = std.Io;
+const Threaded = Io.Threaded;
-const Subscription = struct {
- subject: []const u8,
- client_id: usize,
- sid: []const u8,
+const builtin = @import("builtin");
- fn deinit(self: Subscription, alloc: std.mem.Allocator) void {
- alloc.free(self.subject);
- alloc.free(self.sid);
- }
-};
+const zits = @import("zits");
+const Message = zits.Server.Message;
+const ServerInfo = Message.ServerInfo;
-info: ServerInfo,
-clients: std.AutoHashMapUnmanaged(usize, *Client) = .empty,
+const Server = zits.Server;
-subs_lock: std.Io.Mutex = .init,
-subscriptions: std.ArrayList(Subscription) = .empty,
+const safe_build = builtin.mode == .Debug or builtin.mode == .ReleaseSafe;
-var keep_running = std.atomic.Value(bool).init(true);
+var keep_running = AtomicValue(bool).init(true);
fn handleSigInt(sig: std.os.linux.SIG) callconv(.c) void {
_ = sig;
keep_running.store(false, .monotonic);
}
-pub fn deinit(server: *Server, io: std.Io, alloc: std.mem.Allocator) void {
- server.subs_lock.lockUncancelable(io);
- defer server.subs_lock.unlock(io);
- for (server.subscriptions.items) |sub| {
- sub.deinit(alloc);
- }
- server.subscriptions.deinit(alloc);
- server.clients.deinit(alloc);
-}
-
-pub fn main(alloc: std.mem.Allocator, server_config: ServerInfo) !void {
+pub fn main(outer_alloc: Allocator, server_config: ServerInfo) !void {
// Configure the signal action
- const act = std.posix.Sigaction{
+ const act = Sigaction{
.handler = .{ .handler = handleSigInt },
.mask = std.posix.sigemptyset(),
.flags = 0,
@@ -52,23 +36,24 @@ pub fn main(alloc: std.mem.Allocator, server_config: ServerInfo) !void {
std.posix.sigaction(std.posix.SIG.INT, &act, null);
{
- var dba: std.heap.DebugAllocator(.{}) = .init;
- dba.backing_allocator = alloc;
+ var dba: DebugAllocator(.{}) = .init;
+ dba.backing_allocator = outer_alloc;
defer _ = dba.deinit();
- const gpa = if (builtin.mode == .Debug or builtin.mode == .ReleaseSafe) dba.allocator() else alloc;
+ const alloc = if (safe_build) dba.allocator() else outer_alloc;
- var threaded: std.Io.Threaded = .init(gpa, .{});
+ var threaded: Threaded = .init(alloc, .{});
defer threaded.deinit();
const io = threaded.io();
var server: Server = .{
.info = server_config,
};
- defer server.deinit(io, gpa);
+ defer server.deinit(io, alloc);
- var server_task = try io.concurrent(start, .{ &server, io, gpa });
+ var server_task = try io.concurrent(Server.start, .{ &server, io, alloc });
defer server_task.cancel(io) catch {};
+ // Block until Ctrl+C
while (keep_running.load(.monotonic)) {
try io.sleep(.fromMilliseconds(1), .awake);
}
@@ -79,273 +64,3 @@ pub fn main(alloc: std.mem.Allocator, server_config: ServerInfo) !void {
}
std.log.info("Goodbye", .{});
}
-
-pub fn start(server: *Server, io: std.Io, gpa: std.mem.Allocator) !void {
- var tcp_server = try std.Io.net.IpAddress.listen(try std.Io.net.IpAddress.parse(
- server.info.host,
- server.info.port,
- ), io, .{ .reuse_address = true });
- defer tcp_server.deinit(io);
- std.log.debug("Server headers: {s}", .{if (server.info.headers) "true" else "false"});
- std.log.debug("Server max payload: {d}", .{server.info.max_payload});
- std.log.info("Server ID: {s}", .{server.info.server_id});
- std.log.info("Server name: {s}", .{server.info.server_name});
- std.log.info("Server listening on {s}:{d}", .{ server.info.host, server.info.port });
-
- var client_group: std.Io.Group = .init;
- defer client_group.cancel(io);
-
- var id: usize = 0;
- while (true) : (id +%= 1) {
- if (server.clients.contains(id)) continue;
- std.log.debug("Accepting next client", .{});
- const stream = try tcp_server.accept(io);
- std.log.debug("Accepted connection {d}", .{id});
- _ = client_group.concurrent(io, handleConnectionInfallible, .{ server, gpa, io, id, stream }) catch {
- std.log.err("Could not start concurrent handler for {d}", .{id});
- stream.close(io);
- };
- }
-}
-
-fn addClient(server: *Server, allocator: std.mem.Allocator, id: usize, client: *Client) !void {
- try server.clients.put(allocator, id, client);
-}
-
-fn removeClient(server: *Server, io: std.Io, allocator: std.mem.Allocator, id: usize) void {
- server.subs_lock.lockUncancelable(io);
- defer server.subs_lock.unlock(io);
- if (server.clients.remove(id)) {
- const len = server.subscriptions.items.len;
- for (0..len) |from_end| {
- const i = len - from_end - 1;
- const sub = server.subscriptions.items[i];
- if (sub.client_id == id) {
- sub.deinit(allocator);
- _ = server.subscriptions.swapRemove(i);
- }
- }
- }
-}
-
-fn handleConnectionInfallible(
- server: *Server,
- server_allocator: std.mem.Allocator,
- io: std.Io,
- id: usize,
- stream: std.Io.net.Stream,
-) void {
- handleConnection(server, server_allocator, io, id, stream) catch |err| {
- std.log.err("Failed processing client {d}: {any}", .{ id, err });
- };
-}
-
-fn handleConnection(
- server: *Server,
- server_allocator: std.mem.Allocator,
- io: std.Io,
- id: usize,
- stream: std.Io.net.Stream,
-) !void {
- defer stream.close(io);
-
- //var client_allocator: std.heap.DebugAllocator(.{}) = .init;
- //client_allocator.backing_allocator = server_allocator;
- //defer _ = client_allocator.deinit();
- //const allocator = if (builtin.mode == .Debug or builtin.mode == .ReleaseSafe) client_allocator.allocator() else server_allocator;
-
- // Set up client writer
- var w_buffer: [1024 * 16]u8 = undefined;
- var writer = stream.writer(io, &w_buffer);
- const out = &writer.interface;
-
- // Set up client reader
- var r_buffer: [1024 * 16]u8 = undefined;
- var reader = stream.reader(io, &r_buffer);
- const in = &reader.interface;
-
- // Set up buffer queue
- var qbuf: [8]Message = undefined;
- var queue: std.Io.Queue(Message) = .init(&qbuf);
- defer {
- queue.close(io);
- while (queue.getOne(io)) |msg| {
- switch (msg) {
- .msg => |m| m.deinit(server_allocator),
- .hmsg => |h| h.deinit(server_allocator),
- else => {},
- }
- } else |_| {}
- }
-
- // Create client
- var client: Client = .init(null, &queue, in, out);
- defer client.deinit(server_allocator);
-
- try server.addClient(server_allocator, id, &client);
- defer server.removeClient(io, server_allocator, id);
-
- // Do initial handshake with client
- try queue.putOne(io, .{ .info = server.info });
-
- var client_task = try io.concurrent(Client.start, .{ &client, io, server_allocator });
- defer client_task.cancel(io) catch {};
-
- // Messages are owned by the server after they are received from the client
- while (client.next(server_allocator)) |msg| {
- switch (msg) {
- .ping => {
- // Respond to ping with pong.
- try client.send(io, .pong);
- },
- .@"pub", .hpub => {
- defer switch (msg) {
- .@"pub" => |pb| pb.deinit(server_allocator),
- .hpub => |hp| hp.deinit(server_allocator),
- else => unreachable,
- };
- try server.publishMessage(io, server_allocator, &client, msg);
- },
- .sub => |sub| {
- defer sub.deinit(server_allocator);
- try server.subscribe(io, server_allocator, id, sub);
- },
- .unsub => |unsub| {
- defer unsub.deinit(server_allocator);
- try server.unsubscribe(io, server_allocator, id, unsub);
- },
- .connect => |connect| {
- if (client.connect) |*current| {
- current.deinit(server_allocator);
- }
- client.connect = connect;
- },
- else => |e| {
- std.debug.panic("Unimplemented message: {any}\n", .{e});
- },
- }
- } else |err| switch (err) {
- error.EndOfStream => {
- std.log.debug("Client {d} disconnected", .{id});
- },
- else => {
- return err;
- },
- }
-}
-
-fn subjectMatches(sub_subject: []const u8, pub_subject: []const u8) bool {
- // TODO: assert that sub_subject and pub_subject are valid.
- var sub_iter = std.mem.splitScalar(u8, sub_subject, '.');
- var pub_iter = std.mem.splitScalar(u8, pub_subject, '.');
-
- while (sub_iter.next()) |st| {
- const pt = pub_iter.next() orelse return false;
-
- if (std.mem.eql(u8, st, ">")) return true;
-
- if (!std.mem.eql(u8, st, "*") and !std.mem.eql(u8, st, pt)) {
- return false;
- }
- }
-
- return pub_iter.next() == null;
-}
-
-test subjectMatches {
- try std.testing.expect(subjectMatches("foo", "foo"));
- try std.testing.expect(!subjectMatches("foo", "bar"));
-
- try std.testing.expect(subjectMatches("foo.*", "foo.bar"));
- try std.testing.expect(!subjectMatches("foo.*", "foo"));
- try std.testing.expect(!subjectMatches("foo.>", "foo"));
-
- // the wildcard subscriptions foo.*.quux and foo.> both match foo.bar.quux, but only the latter matches foo.bar.baz.
- try std.testing.expect(subjectMatches("foo.*.quux", "foo.bar.quux"));
- try std.testing.expect(subjectMatches("foo.>", "foo.bar.quux"));
- try std.testing.expect(!subjectMatches("foo.*.quux", "foo.bar.baz"));
- try std.testing.expect(subjectMatches("foo.>", "foo.bar.baz"));
-}
-
-fn publishMessage(server: *Server, io: std.Io, alloc: std.mem.Allocator, source_client: *Client, msg: Message) !void {
- errdefer {
- if (source_client.connect) |c| {
- if (c.verbose) {
- source_client.send(io, .{ .@"-err" = "Slow Consumer" }) catch {};
- }
- }
- }
- const subject = switch (msg) {
- .@"pub" => |pb| pb.subject,
- .hpub => |hp| hp.@"pub".subject,
- else => unreachable,
- };
- try server.subs_lock.lock(io);
- defer server.subs_lock.unlock(io);
- for (server.subscriptions.items) |subscription| {
- if (subjectMatches(subscription.subject, subject)) {
- const client = server.clients.get(subscription.client_id) orelse {
- std.debug.print("trying to publish to a client that no longer exists: {d}\n", .{subscription.client_id});
- continue;
- };
-
- switch (msg) {
- .@"pub" => |pb| client.send(io, .{
- .msg = try pb.toMsg(alloc, subscription.sid),
- }) catch |err| switch (err) {
- error.Canceled => return err,
- else => {},
- },
- .hpub => |hp| client.send(io, .{ .hmsg = try hp.toHMsg(
- alloc,
- subscription.sid,
- ) }) catch |err| switch (err) {
- error.Canceled => return err,
- else => {},
- },
- else => unreachable,
- }
- }
- }
- if (source_client.connect) |c| {
- if (c.verbose) {
- source_client.send(io, .@"+ok") catch {};
- }
- }
-}
-
-fn subscribe(server: *Server, io: std.Io, gpa: std.mem.Allocator, id: usize, msg: Message.Sub) !void {
- try server.subs_lock.lock(io);
- defer server.subs_lock.unlock(io);
- const subject = try gpa.dupe(u8, msg.subject);
- errdefer gpa.free(subject);
- const sid = try gpa.dupe(u8, msg.sid);
- errdefer gpa.free(sid);
- try server.subscriptions.append(gpa, .{
- .subject = subject,
- .client_id = id,
- .sid = sid,
- });
-}
-
-fn unsubscribe(server: *Server, io: std.Io, gpa: std.mem.Allocator, id: usize, msg: Message.Unsub) !void {
- try server.subs_lock.lock(io);
- defer server.subs_lock.unlock(io);
- const len = server.subscriptions.items.len;
- for (0..len) |from_end| {
- const i = len - from_end - 1;
- const sub = server.subscriptions.items[i];
- if (sub.client_id == id and std.mem.eql(u8, sub.sid, msg.sid)) {
- sub.deinit(gpa);
- _ = server.subscriptions.swapRemove(i);
- }
- }
-}
-
-pub fn createId() []const u8 {
- return "SERVERID";
-}
-
-pub fn createName() []const u8 {
- return "SERVERNAME";
-}
diff --git a/src/server/test.zig b/src/server/test.zig
deleted file mode 100644
index e89e354..0000000
--- a/src/server/test.zig
+++ /dev/null
@@ -1,3 +0,0 @@
-const std = @import("std");
-const Server = @import("./main.zig");
-const Client = @import("./client.zig").ClientState;
generated by cgit (git 2.34.1) at 2026-02-02 14:08:01 +0000