diff options
Diffstat (limited to 'src/server/main.zig')
| -rw-r--r-- | src/server/main.zig | 33 |
1 files changed, 16 insertions, 17 deletions
diff --git a/src/server/main.zig b/src/server/main.zig index 044f551..aad842c 100644 --- a/src/server/main.zig +++ b/src/server/main.zig @@ -87,7 +87,6 @@ pub fn start(server: *Server, io: std.Io, gpa: std.mem.Allocator) !void { defer client_group.cancel(io); var id: usize = 0; - // Run until SIGINT is handled, then exit gracefully while (true) : (id +%= 1) { std.debug.print("in server loop\n", .{}); if (server.clients.contains(id)) continue; @@ -140,10 +139,10 @@ fn handleConnection( ) !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; + //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]u8 = undefined; @@ -157,6 +156,8 @@ fn handleConnection( // Create client var client: Client = .init(null, in, out); + defer client.deinit(server_allocator); + try server.addClient(server_allocator, id, &client); defer server.removeClient(io, server_allocator, id); @@ -172,17 +173,9 @@ fn handleConnection( } else |_| {} } - var client_task = try io.concurrent(Client.start, .{ &client, io, server_allocator, &queue }); + var client_task = try io.concurrent(Client.start, .{ &client, io, server_allocator, &queue, server.info }); defer client_task.cancel(io) catch {}; - try io.sleep(std.Io.Duration.fromMilliseconds(5), .real); - - // Do initial handshake with client - try client.send(io, .{ .info = server.info }); - var connect_arena: std.heap.ArenaAllocator = .init(allocator); - defer connect_arena.deinit(); - client.connect = (Message.next(connect_arena.allocator(), in) catch return).connect; - // Messages are owned by the server after they are received from the client while (client.next(server_allocator)) |msg| { switch (msg) { @@ -191,6 +184,7 @@ fn handleConnection( try client.send(io, .pong); }, .@"pub" => |pb| { + defer pb.deinit(server_allocator); try server.publishMessage(io, server_allocator, &client, pb); }, .sub => |sub| { @@ -199,6 +193,12 @@ fn handleConnection( .unsub => |unsub| { 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}); }, @@ -213,8 +213,8 @@ fn handleConnection( } } -fn subjectMatches(expected: []const u8, actual: []const u8) bool { - return std.mem.eql(u8, expected, actual); +fn subjectMatches(sub_subject: []const u8, pub_subject: []const u8) bool { + return std.mem.eql(u8, sub_subject, pub_subject); } fn publishMessage(server: *Server, io: std.Io, alloc: std.mem.Allocator, source_client: *Client, msg: Message.Pub) !void { @@ -225,7 +225,6 @@ fn publishMessage(server: *Server, io: std.Io, alloc: std.mem.Allocator, source_ } } } - defer msg.deinit(alloc); try server.subs_lock.lock(io); defer server.subs_lock.unlock(io); for (server.subscriptions.items) |subscription| { |