From 1eeb55ff4defe4bfaeaadaa62146604562ee0ea4 Mon Sep 17 00:00:00 2001
From: Robby Zambito <contact@robbyzambito.me>
Date: Mon, 8 Dec 2025 21:09:15 -0500
Subject: Made progress, but not perfect.

the message isn't crossing container boundaries, but it works in the
test!
---
 src/server/main.zig | 41 +++++++++++++++++++++++++----------------
 1 file changed, 25 insertions(+), 16 deletions(-)

(limited to 'src/server/main.zig')

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| {
-- 
cgit