diff options
Diffstat (limited to 'src/server/main.zig')
| -rw-r--r-- | src/server/main.zig | 41 |
1 files changed, 25 insertions, 16 deletions
diff --git a/src/server/main.zig b/src/server/main.zig index 7f9b9a3..155a455 100644 --- a/src/server/main.zig +++ b/src/server/main.zig @@ -6,7 +6,7 @@ const ClientState = @import("./client.zig").ClientState; const Server = @This(); info: ServerInfo, -clients: std.AutoHashMapUnmanaged(usize, ClientState) = .empty, +clients: std.AutoHashMapUnmanaged(usize, *ClientState) = .empty, /// Map of subjects to a map of (client ID => SID) subscriptions: std.StringHashMapUnmanaged(std.AutoHashMapUnmanaged(usize, []const u8)) = .empty, @@ -27,8 +27,10 @@ pub fn main(gpa: std.mem.Allocator, server_config: ServerInfo) !void { var id: usize = 0; while (true) : (id +%= 1) { + std.debug.print("in server loop\n", .{}); if (server.clients.contains(id)) continue; const stream = try tcp_server.accept(io); + std.debug.print("accepted connection\n", .{}); _ = io.concurrent(handleConnection, .{ &server, gpa, io, id, stream }) catch { std.debug.print("could not start concurrent handler for {d}\n", .{id}); stream.close(io); @@ -36,7 +38,7 @@ pub fn main(gpa: std.mem.Allocator, server_config: ServerInfo) !void { } } -fn addClient(server: *Server, allocator: std.mem.Allocator, id: usize, client: ClientState) !void { +fn addClient(server: *Server, allocator: std.mem.Allocator, id: usize, client: *ClientState) !void { // server.clients.lockPointers(); try server.clients.put(allocator, id, client); // server.clients.unlockPointers(); @@ -71,7 +73,7 @@ fn handleConnection( const connect = (Message.next(connect_arena.allocator(), in) catch return).connect; var client_state: ClientState = try .init(io, allocator, id, connect, in, out); - try server.addClient(allocator, id, client_state); + try server.addClient(allocator, id, &client_state); defer server.removeClient(allocator, id); // defer { @@ -113,12 +115,12 @@ fn publishMessage(server: *Server, io: std.Io, msg: Message.Pub) !void { const client_id = sub.key_ptr.*; const sid = sub.value_ptr.*; - var client = server.clients.getPtr(client_id) orelse { + const client = server.clients.getPtr(client_id) orelse { std.debug.print("trying to publish to a client that no longer exists: {d}", .{client_id}); continue; }; - _ = try client.send(io, .{ .msg = .{ + _ = try client.*.send(io, .{ .msg = .{ .subject = msg.subject, .sid = sid, .reply_to = msg.reply_to, @@ -141,9 +143,10 @@ pub fn processClient(server: *Server, gpa: std.mem.Allocator, io: std.Io, client defer client_state.deinit(gpa); std.debug.print("processing client: {d}\n", .{client_state.id}); - while (true) { - std.debug.print("awaiting next message from client\n", .{}); - switch (client_state.next(io)) { + std.debug.print("awaiting next message from client\n", .{}); + while (client_state.next(io)) |msg| { + std.debug.print("message from client!: {any}\n", .{msg}); + switch (msg) { .ping => { std.debug.print("got a ping! sending a pong.\n", .{}); for (0..5) |_| { @@ -156,22 +159,28 @@ pub fn processClient(server: *Server, gpa: std.mem.Allocator, io: std.Io, client std.debug.print("could not pong to client {d}\n", .{client_state.id}); } }, - .@"pub" => |msg| { - try server.publishMessage(io, msg); - if (client_state.connect.verbose) { + .@"pub" => |@"pub"| { + std.debug.print("pub: {any}\n", .{@"pub"}); + try server.publishMessage(io, @"pub"); + if (client_state.connect.connect.verbose) { _ = try client_state.send(io, .@"+ok"); } }, - .sub => |msg| { - try server.subscribe(gpa, client_state.id, msg); + .sub => |sub| { + try server.subscribe(gpa, client_state.id, sub); + }, + .eos => { + client_state.io_group.wait(io); + break; }, - else => |msg| { - std.debug.panic("Unimplemented message: {any}\n", .{msg}); + else => |e| { + std.debug.panic("Unimplemented message: {any}\n", .{e}); }, } std.debug.print("processed message from client\n", .{}); - } + std.debug.print("awaiting next message from client\n", .{}); + } else |_| {} // while (!io.cancelRequested()) { // if (client_state.send_queue.getOne(io)) |msg| { |