The issue was the writer is not working to send pong ??

src/server/client.zig

@@ -4,21 +4,18 @@ const std = @import("std");
 pub const ClientState = struct {
     id: usize,
     connect: Message.AllocatedConnect,
-    /// Messages that this client is trying to send.
+
-    send_queue: std.Io.Queue(Message),
-    send_queue_buffer: [1024]Message = undefined,
     /// Messages that this client should receive.
     recv_queue: std.Io.Queue(Message),
     recv_queue_buffer: [1024]Message = undefined,
 
-    io_group: std.Io.Group = .init,
-
     from_client: *std.Io.Reader,
     to_client: *std.Io.Writer,
 
+    task: std.Io.Future(void),
+
     pub fn init(
         io: std.Io,
-        allocator: std.mem.Allocator,
         id: usize,
         connect: Message.AllocatedConnect,
         in: *std.Io.Reader,
@@ -27,15 +24,13 @@ pub const ClientState = struct {
         var res: ClientState = .{
             .id = id,
             .connect = connect,
-            .send_queue = undefined,
             .recv_queue = undefined,
             .from_client = in,
             .to_client = out,
+            .task = undefined,
         };
-        res.send_queue = .init(&res.send_queue_buffer);
         res.recv_queue = .init(&res.recv_queue_buffer);
-        try res.io_group.concurrent(io, processWrite, .{ &res, io, out });
+        res.task = try io.concurrent(processWrite, .{ &res, io, out });
-        try res.io_group.concurrent(io, processRead, .{ &res, io, allocator, in });
 
         return res;
     }
@@ -47,6 +42,7 @@ pub const ClientState = struct {
     ) void {
         while (true) {
             const message = self.recv_queue.getOne(io) catch continue;
+            std.debug.print("got message in write loop to send to client: {any}\n", .{message});
             switch (message) {
                 .@"+ok" => {
                     writeOk(out) catch break;
@@ -67,63 +63,23 @@ pub const ClientState = struct {
         }
     }
 
-    fn processRead(
+    pub fn deinit(self: *ClientState, io: std.Io) void {
-        self: *ClientState,
+        self.task.cancel(io);
-        io: std.Io,
+        self.connect.deinit();
-        allocator: std.mem.Allocator,
-        in: *std.Io.Reader,
-    ) void {
-        // io.sleep(.fromMilliseconds(100), .real) catch @panic("couldn't sleep");
-        while (true) {
-            // std.debug.print("waiting for message\n", .{});
-            const next_message = Message.next(allocator, in) catch |err| switch (err) {
-                error.EndOfStream => {
-                    self.send_queue.putOne(io, .eos) catch {};
-                    break;
-                },
-                else => {
-                    std.debug.panic("guh: {any}\n", .{err});
-                    break;
-                },
-            };
-            std.debug.print("next_message: {any}\n", .{next_message});
-
-            switch (next_message) {
-                .eos => {
-                    self.send_queue.putOne(io, next_message) catch {};
-                    break;
-                },
-                else => {
-                    self.send_queue.putOne(io, next_message) catch break;
-                },
-            }
-        }
-
-        std.debug.print("no more messages\n", .{});
-    }
-
-    pub fn deinit(self: *ClientState, alloc: std.mem.Allocator) void {
-        _ = self;
-        // self.write_task.cancel() catch {
-        //     std.debug.print("failed to cancel write task of {}\n", .{self.id});
-        // };
-        // self.read_taks.cancel() catch {
-        //     std.debug.print("failed to cancel read task of {}\n", .{self.id});
-        // };
-        // TODO: dealloc things.
-        _ = alloc;
     }
 
     /// Return true if the value was put in the clients buffer to process, else false.
     pub fn send(self: *ClientState, io: std.Io, msg: Message) std.Io.Cancelable!bool {
-        return (try self.recv_queue.put(io, &.{msg}, 0)) > 0;
+        try self.recv_queue.putOne(io, msg);
+        return true;
     }
 
-    pub fn next(self: *ClientState, io: std.Io) !Message {
+    pub fn next(self: *ClientState, allocator: std.mem.Allocator) !Message {
-        std.debug.print("in client awaiting next message\n", .{});
+        // std.debug.print("in client awaiting next message\n", .{});
-        errdefer std.debug.print("actually it was canceled\n", .{});
+        // errdefer std.debug.print("actually it was canceled\n", .{});
-        defer std.debug.print("client returning next message!\n", .{});
+        // defer std.debug.print("client returning next message!\n", .{});
-        return self.send_queue.getOne(io);
+        return Message.next(allocator, self.from_client);
+        // return self.send_queue.getOne(io);
     }
 };
 
@@ -132,11 +88,13 @@ fn writeOk(out: *std.Io.Writer) !void {
     try out.flush();
 }
 
-fn writePong(out: *std.Io.Writer) !void {
+pub fn writePong(out: *std.Io.Writer) !void {
     std.debug.print("writing pong\n", .{});
+    for (0..10000) |_| {
         _ = try out.write("PONG\r\n");
         try out.flush();
     }
+}
 
 pub fn writeInfo(out: *std.Io.Writer, info: Message.ServerInfo) !void {
     std.debug.print("writing info: {any}\n", .{info});

src/server/main.zig

@@ -71,7 +71,8 @@ fn handleConnection(
     var connect_arena: std.heap.ArenaAllocator = .init(allocator);
     defer connect_arena.deinit();
     const connect = (Message.next(connect_arena.allocator(), in) catch return).connect;
-    var client_state: ClientState = try .init(io, allocator, id, connect, in, out);
+    var client_state: ClientState = try .init(io, id, connect, in, out);
+    defer client_state.deinit(io);
 
     try server.addClient(allocator, id, &client_state);
     defer server.removeClient(allocator, id);
@@ -96,59 +97,17 @@ fn handleConnection(
     //     server.subscriptions.unlockPointers();
     // }
 
-    server.processClient(allocator, io, &client_state) catch |err| {
+    {
-        std.debug.panic("Error processing client: {}\n", .{err});
-    };
-}
-
-// // Result is owned by the caller
-// fn subscribers(server: *Server, gpa: std.mem.Allocator, subject: []const u8) []ClientState {
-//     var acc: std.ArrayList(ClientState) = .empty;
-
-//     return acc.toOwnedSlice();
-// }
-
-fn publishMessage(server: *Server, io: std.Io, msg: Message.Pub) !void {
-    if (server.subscriptions.get(msg.subject)) |subs| {
-        var subs_iter = subs.iterator();
-        while (subs_iter.next()) |sub| {
-            const client_id = sub.key_ptr.*;
-            const sid = sub.value_ptr.*;
-
-            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 = .{
-                .subject = msg.subject,
-                .sid = sid,
-                .reply_to = msg.reply_to,
-                .payload = msg.payload,
-            } });
-        }
-    } else {
-        std.debug.print("no subs on {s}\n", .{msg.subject});
-    }
-}
-
-fn subscribe(server: *Server, gpa: std.mem.Allocator, id: usize, msg: Message.Sub) !void {
-    var subs_for_subject: std.AutoHashMapUnmanaged(usize, []const u8) = if (server.subscriptions.fetchRemove(msg.subject)) |s| s.value else .empty;
-    try subs_for_subject.put(gpa, id, msg.sid);
-    try server.subscriptions.put(gpa, msg.subject, subs_for_subject);
-}
-
-pub fn processClient(server: *Server, gpa: std.mem.Allocator, io: std.Io, client_state: *ClientState) !void {
         defer std.debug.print("done processing client??\n", .{});
-    defer client_state.deinit(gpa);
         std.debug.print("processing client: {d}\n", .{client_state.id});
 
         std.debug.print("awaiting next message from client\n", .{});
-    while (client_state.next(io)) |msg| {
+        while (client_state.next(allocator)) |msg| {
             std.debug.print("message from client!: {any}\n", .{msg});
             switch (msg) {
                 .ping => {
                     std.debug.print("got a ping! sending a pong.\n", .{});
+                    @import("./client.zig").writePong(out) catch return;
                     for (0..5) |_| {
                         if (try client_state.send(io, .pong)) {
                             std.debug.print("sent pong\n", .{});
@@ -163,14 +122,16 @@ pub fn processClient(server: *Server, gpa: std.mem.Allocator, io: std.Io, client
                     std.debug.print("pub: {any}\n", .{@"pub"});
                     try server.publishMessage(io, @"pub");
                     if (client_state.connect.connect.verbose) {
+                        std.debug.print("server IS sending +ok\n", .{});
                         _ = try client_state.send(io, .@"+ok");
+                    } else {
+                        std.debug.print("server NOT sending +ok\n", .{});
                     }
                 },
                 .sub => |sub| {
-                try server.subscribe(gpa, client_state.id, sub);
+                    try server.subscribe(allocator, client_state.id, sub);
                 },
                 .eos => {
-                client_state.io_group.wait(io);
                     break;
                 },
                 else => |e| {
@@ -262,6 +223,46 @@ pub fn processClient(server: *Server, gpa: std.mem.Allocator, io: std.Io, client
         // }
     }
 
+    client_state.task.await(io);
+}
+
+// // Result is owned by the caller
+// fn subscribers(server: *Server, gpa: std.mem.Allocator, subject: []const u8) []ClientState {
+//     var acc: std.ArrayList(ClientState) = .empty;
+
+//     return acc.toOwnedSlice();
+// }
+
+fn publishMessage(server: *Server, io: std.Io, msg: Message.Pub) !void {
+    if (server.subscriptions.get(msg.subject)) |subs| {
+        var subs_iter = subs.iterator();
+        while (subs_iter.next()) |sub| {
+            const client_id = sub.key_ptr.*;
+            const sid = sub.value_ptr.*;
+
+            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 = .{
+                .subject = msg.subject,
+                .sid = sid,
+                .reply_to = msg.reply_to,
+                .payload = msg.payload,
+            } });
+        }
+    } else {
+        std.debug.print("no subs on {s}\n", .{msg.subject});
+    }
+}
+
+fn subscribe(server: *Server, gpa: std.mem.Allocator, id: usize, msg: Message.Sub) !void {
+    var subs_for_subject: std.AutoHashMapUnmanaged(usize, []const u8) = if (server.subscriptions.fetchRemove(msg.subject)) |s| s.value else .empty;
+    try subs_for_subject.put(gpa, id, msg.sid);
+    try server.subscriptions.put(gpa, msg.subject, subs_for_subject);
+}
+
 pub fn createId() []const u8 {
     return "SERVERID";
 }