1 files changed, 34 insertions, 23 deletions

diff --git a/src/server/client.zig b/src/server/client.zig
index 44de542..684a50f 100644
--- a/src/server/client.zig
+++ b/src/server/client.zig
@@ -5,7 +5,8 @@ const Client = @This();
 
 connect: ?Message.Connect,
 
-write_lock: std.Io.Mutex,
+// Messages for this client to receive.
+recv_queue: ?*std.Io.Queue(Message) = null,
 
 from_client: *std.Io.Reader,
 to_client: *std.Io.Writer,
@@ -17,36 +18,46 @@ pub fn init(
 ) Client {
     return .{
         .connect = connect,
-        .write_lock = .init,
         .from_client = in,
         .to_client = out,
     };
 }
 
-/// Return true if the value was put in the clients buffer to process, else false.
-pub fn send(self: *Client, io: std.Io, msg: Message) !void {
-    try self.write_lock.lock(io);
-    defer self.write_lock.unlock(io);
-
-    switch (msg) {
-        .@"+ok" => {
-            try writeOk(self.to_client);
-        },
-        .pong => {
-            try writePong(self.to_client);
-        },
-        .info => |info| {
-            try writeInfo(self.to_client, info);
-        },
-        .msg => |m| {
-            try writeMsg(self.to_client, m);
-        },
-        else => {
-            std.debug.panic("unimplemented write", .{});
-        },
+pub fn start(self: *Client, io: std.Io, alloc: std.mem.Allocator, queue: *std.Io.Queue(Message)) !void {
+    self.recv_queue = queue;
+    var msgs: [16]Message = undefined;
+    while (true) {
+        const len = try queue.get(io, &msgs, 1);
+        std.debug.assert(len <= msgs.len);
+        for (msgs[0..len]) |msg| {
+            switch (msg) {
+                .@"+ok" => {
+                    try writeOk(self.to_client);
+                },
+                .pong => {
+                    try writePong(self.to_client);
+                },
+                .info => |info| {
+                    try writeInfo(self.to_client, info);
+                },
+                .msg => |m| {
+                    defer m.deinit(alloc);
+                    try writeMsg(self.to_client, m);
+                },
+                else => |m| {
+                    std.debug.panic("unimplemented write: {any}\n", .{m});
+                },
+            }
+        }
     }
 }
 
+pub fn send(self: *Client, io: std.Io, msg: Message) !void {
+    if (self.recv_queue) |queue| {
+        try queue.putOne(io, msg);
+    } else @panic("Must start() the client before sending it messages.");
+}
+
 pub fn next(self: *Client, allocator: std.mem.Allocator) !Message {
     // std.debug.print("in client awaiting next message\n", .{});
     // errdefer std.debug.print("actually it was canceled\n", .{});