1 files changed, 97 insertions, 18 deletions

diff --git a/src/server/client.zig b/src/server/client.zig
index c8a9239..b4bc55b 100644
--- a/src/server/client.zig
+++ b/src/server/client.zig
@@ -24,7 +24,7 @@ pub const ClientState = struct {
         connect: Message.Connect,
         in: *std.Io.Reader,
         out: *std.Io.Writer,
-    ) ClientState {
+    ) !ClientState {
         var res: ClientState = .{
             .id = id,
             .connect = connect,
@@ -37,12 +37,10 @@ pub const ClientState = struct {
         };
         res.send_queue = .init(&res.send_queue_buffer);
         res.recv_queue = .init(&res.recv_queue_buffer);
-        const write_task = io.async(processWrite, .{ &res, io, out });
-        // @compileLog(@TypeOf(write_task));
-        const read_task = io.async(processRead, .{ &res, io, allocator, in });
-        // @compileLog(@TypeOf(read_task));
-        res.write_task = write_task;
-        res.read_task = read_task;
+        // res.send_queue = .init(&.{});
+        // res.recv_queue = .init(&.{});
+        res.write_task = try io.concurrent(processWrite, .{ &res, io, out });
+        res.read_task = try io.concurrent(processRead, .{ &res, io, allocator, in });
 
         return res;
     }
@@ -53,13 +51,23 @@ pub const ClientState = struct {
         out: *std.Io.Writer,
     ) void {
         while (true) {
-            const message = self.recv_queue.getOne(io) catch break;
+            const message = self.recv_queue.getOne(io) catch continue;
             switch (message) {
-                .@"+ok" => writeOk(out) catch break,
-                .pong => writePong(out) catch break,
-                .info => |info| writeInfo(out, info) catch break,
-                .msg => |m| writeMsg(out, m) catch break,
-                else => std.debug.panic("unimplemented write", .{}),
+                .@"+ok" => {
+                    writeOk(out) catch break;
+                },
+                .pong => {
+                    writePong(out) catch break;
+                },
+                .info => |info| {
+                    writeInfo(out, info) catch break;
+                },
+                .msg => |m| {
+                    writeMsg(out, m) catch break;
+                },
+                else => {
+                    std.debug.panic("unimplemented write", .{});
+                },
             }
         }
     }
@@ -70,19 +78,27 @@ pub const ClientState = struct {
         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 => {
-                    break;
-                },
+                error.EndOfStream => break,
                 else => {
                     std.debug.panic("guh: {any}\n", .{err});
                     break;
                     // return err;
                 },
             };
-            self.send_queue.putOne(io, next_message) catch break;
+            std.debug.print("got message {}\n", .{next_message});
+            // std.debug.print("queue: {any}\n", .{self.send_queue});
+            self.send_queue.putOneUncancelable(io, next_message); //catch {
+            //     std.debug.print("in catch\n\n\n", .{});
+
+            // std.debug.print("queue: {any}\n", .{self.send_queue});
+            // };
         }
+
+        std.debug.print("no more messages\n", .{});
     }
 
     pub fn deinit(self: *ClientState, alloc: std.mem.Allocator) void {
@@ -102,7 +118,10 @@ pub const ClientState = struct {
         return (try self.recv_queue.put(io, &.{msg}, 0)) > 0;
     }
 
-    pub fn next(self: *ClientState, io: std.Io) std.Io.Cancelable!Message {
+    pub fn next(self: *ClientState, io: std.Io) !Message {
+        std.debug.print("in client awaiting next message\n", .{});
+        errdefer std.debug.print("actually it was canceled\n", .{});
+        defer std.debug.print("client returning next message!\n", .{});
         return self.send_queue.getOne(io);
     }
 };
@@ -113,11 +132,13 @@ fn writeOk(out: *std.Io.Writer) !void {
 }
 
 fn writePong(out: *std.Io.Writer) !void {
+    std.debug.print("writing pong\n", .{});
     _ = 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});
     _ = try out.write("INFO ");
     try std.json.Stringify.value(info, .{}, out);
     _ = try out.write("\r\n");
@@ -138,3 +159,61 @@ fn writeMsg(out: *std.Io.Writer, msg: Message.Msg) !void {
     );
     try out.flush();
 }
+
+test {
+    const io = std.testing.io;
+    const gpa = std.testing.allocator;
+
+    var from_client: std.Io.Reader = .fixed(
+        "CONNECT {\"verbose\":false,\"pedantic\":false,\"tls_required\":false,\"name\":\"NATS CLI Version v0.2.4\",\"lang\":\"go\",\"version\":\"1.43.0\",\"protocol\":1,\"echo\":true,\"headers\":true,\"no_responders\":true}\r\n" ++
+            "PING\r\n",
+    );
+    var from_client_buf: [1024]Message = undefined;
+    var from_client_queue: std.Io.Queue(Message) = .init(&from_client_buf);
+
+    while (Message.next(gpa, &from_client)) |msg| {
+        try from_client_queue.putOne(io, msg);
+    } else |_| {}
+
+    for (0..2) |_| {
+        var msg = try from_client_queue.getOne(io);
+        std.debug.print("Message: {any}\n", .{msg});
+        switch (msg) {
+            .connect => |*c| {
+                c.deinit();
+            },
+            else => {},
+        }
+    }
+
+    // const connect = (Message.next(gpa, &from_client) catch unreachable).connect;
+
+    // var to_client_alloc: std.Io.Writer.Allocating = .init(gpa);
+    // defer to_client_alloc.deinit();
+    // var to_client = to_client_alloc.writer;
+
+    // var client: ClientState = try .init(io, gpa, 0, connect, &from_client, &to_client);
+    // defer client.deinit(gpa);
+
+    // {
+    //     var get_next = io.concurrent(ClientState.next, .{ &client, io }) catch unreachable;
+    //     defer if (get_next.cancel(io)) |_| {} else |_| @panic("fail");
+
+    //     var timeout = io.concurrent(std.Io.sleep, .{ io, .fromMilliseconds(1000), .awake }) catch unreachable;
+    //     defer timeout.cancel(io) catch {};
+
+    //     switch (try io.select(.{
+    //         .get_next = &get_next,
+    //         .timeout = &timeout,
+    //     })) {
+    //         .get_next => |next| {
+    //             std.debug.print("next is {any}\n", .{next});
+    //             try std.testing.expect((next catch |err| return err) == .ping);
+    //         },
+    //         .timeout => {
+    //             std.debug.print("reached timeout\n", .{});
+    //             return error.TestUnexpectedResult;
+    //         },
+    //     }
+    // }
+}