sending errors on pub sub!

1 files changed, 16 insertions, 14 deletions

diff --git a/src/server/client.zig b/src/server/client.zig
index 05f7859..d6ccacf 100644
--- a/src/server/client.zig
+++ b/src/server/client.zig
@@ -6,7 +6,7 @@ pub const ClientState = struct {
     connect: Message.AllocatedConnect,
 
     /// Messages that this client should receive.
-    recv_queue: std.Io.Queue(Message),
+    recv_queue: *std.Io.Queue(Message),
     recv_queue_buffer: [1024]Message = undefined,
 
     from_client: *std.Io.Reader,
@@ -16,6 +16,7 @@ pub const ClientState = struct {
 
     pub fn init(
         io: std.Io,
+        allocator: std.mem.Allocator,
         id: usize,
         connect: Message.AllocatedConnect,
         in: *std.Io.Reader,
@@ -24,13 +25,13 @@ pub const ClientState = struct {
         var res: ClientState = .{
             .id = id,
             .connect = connect,
-            .recv_queue = undefined,
+            .recv_queue = try allocator.create(std.Io.Queue(Message)),
             .from_client = in,
             .to_client = out,
             .task = undefined,
         };
-        res.recv_queue = .init(&res.recv_queue_buffer);
-        res.task = try io.concurrent(processWrite, .{ &res, io, out });
+        res.recv_queue.* = .init(&res.recv_queue_buffer);
+        res.task = try io.concurrent(processWrite, .{ &res, io });
 
         return res;
     }
@@ -38,23 +39,24 @@ pub const ClientState = struct {
     fn processWrite(
         self: *ClientState,
         io: std.Io,
-        out: *std.Io.Writer,
     ) void {
+        std.debug.print("out pointer in write proc: {*}\n", .{self.to_client});
+        std.debug.print("recv queue in write proc: {*}\n", .{&self.recv_queue});
         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;
+                    writeOk(self.to_client) catch break;
                 },
                 .pong => {
-                    writePong(out) catch break;
+                    writePong(self.to_client) catch break;
                 },
                 .info => |info| {
-                    writeInfo(out, info) catch break;
+                    writeInfo(self.to_client, info) catch break;
                 },
                 .msg => |m| {
-                    writeMsg(out, m) catch break;
+                    writeMsg(self.to_client, m) catch break;
                 },
                 else => {
                     std.debug.panic("unimplemented write", .{});
@@ -63,9 +65,10 @@ pub const ClientState = struct {
         }
     }
 
-    pub fn deinit(self: *ClientState, io: std.Io) void {
+    pub fn deinit(self: *ClientState, io: std.Io, allocator: std.mem.Allocator) void {
         self.task.cancel(io);
         self.connect.deinit();
+        allocator.destroy(self.recv_queue);
     }
 
     /// Return true if the value was put in the clients buffer to process, else false.
@@ -89,11 +92,10 @@ fn writeOk(out: *std.Io.Writer) !void {
 }
 
 pub fn writePong(out: *std.Io.Writer) !void {
+    std.debug.print("out pointer: {*}\n", .{out});
     std.debug.print("writing pong\n", .{});
-    for (0..10000) |_| {
-        _ = try out.write("PONG\r\n");
-        try out.flush();
-    }
+    _ = try out.write("PONG\r\n");
+    try out.flush();
 }
 
 pub fn writeInfo(out: *std.Io.Writer, info: Message.ServerInfo) !void {