1 files changed, 108 insertions, 19 deletions

diff --git a/src/server/client.zig b/src/server/client.zig
index 037a4fa..c8a9239 100644
--- a/src/server/client.zig
+++ b/src/server/client.zig
@@ -11,41 +11,130 @@ pub const ClientState = struct {
     recv_queue: std.Io.Queue(Message),
     recv_queue_buffer: [1024]Message = undefined,
 
-    /// Mapping of the subjects to the IDs.
-    subscription_ids: std.StringHashMapUnmanaged([]const u8) = .empty,
+    from_client: *std.Io.Reader,
+    to_client: *std.Io.Writer,
 
-    pub fn init(id: usize, connect: Message.Connect) ClientState {
+    write_task: std.Io.Future(void),
+    read_task: std.Io.Future(void),
+
+    pub fn init(
+        io: std.Io,
+        allocator: std.mem.Allocator,
+        id: usize,
+        connect: Message.Connect,
+        in: *std.Io.Reader,
+        out: *std.Io.Writer,
+    ) ClientState {
         var res: ClientState = .{
             .id = id,
             .connect = connect,
             .send_queue = undefined,
             .recv_queue = undefined,
+            .from_client = in,
+            .to_client = out,
+            .write_task = undefined,
+            .read_task = undefined,
         };
         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;
+
         return res;
     }
 
+    fn processWrite(
+        self: *ClientState,
+        io: std.Io,
+        out: *std.Io.Writer,
+    ) void {
+        while (true) {
+            const message = self.recv_queue.getOne(io) catch break;
+            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", .{}),
+            }
+        }
+    }
+
+    fn processRead(
+        self: *ClientState,
+        io: std.Io,
+        allocator: std.mem.Allocator,
+        in: *std.Io.Reader,
+    ) void {
+        while (true) {
+            const next_message = Message.next(allocator, in) catch |err| switch (err) {
+                error.EndOfStream => {
+                    break;
+                },
+                else => {
+                    std.debug.panic("guh: {any}\n", .{err});
+                    break;
+                    // return err;
+                },
+            };
+            self.send_queue.putOne(io, next_message) catch break;
+        }
+    }
+
     pub fn deinit(self: *ClientState, alloc: std.mem.Allocator) void {
-        self.subscription_ids.clearAndFree(alloc);
+        _ = 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;
     }
 
-    pub fn subscribe(self: *ClientState, gpa: std.mem.Allocator, sub: Message.Sub) !void {
-        self.subscription_ids.lockPointers();
-        defer self.subscription_ids.unlockPointers();
-        try self.subscription_ids.putNoClobber(gpa, sub.subject, sub.sid);
+    /// 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;
     }
 
-    pub fn unsubscribe(self: *ClientState, gpa: std.mem.Allocator, sid: []const u8) void {
-        _ = gpa;
-        self.subscription_ids.lockPointers();
-        defer self.subscription_ids.unlockPointers();
-        var iter = self.subscription_ids.iterator();
-        while (iter.next()) |entry| {
-            if (std.mem.eql(u8, sid, entry.value_ptr.*)) {
-                self.subscription_ids.swapRemove(entry.value_ptr.*);
-                break;
-            }
-        } else unreachable; // Assert that the SID is in the subscriptions.
+    pub fn next(self: *ClientState, io: std.Io) std.Io.Cancelable!Message {
+        return self.send_queue.getOne(io);
     }
 };
+
+fn writeOk(out: *std.Io.Writer) !void {
+    _ = try out.write("+OK\r\n");
+    try out.flush();
+}
+
+fn writePong(out: *std.Io.Writer) !void {
+    _ = try out.write("PONG\r\n");
+    try out.flush();
+}
+
+pub fn writeInfo(out: *std.Io.Writer, info: Message.ServerInfo) !void {
+    _ = try out.write("INFO ");
+    try std.json.Stringify.value(info, .{}, out);
+    _ = try out.write("\r\n");
+    try out.flush();
+}
+
+fn writeMsg(out: *std.Io.Writer, msg: Message.Msg) !void {
+    std.debug.print("PRINTING MESSAGE\n\n\n\n", .{});
+    try out.print(
+        "MSG {s} {s} {s} {d}\r\n{s}\r\n",
+        .{
+            msg.subject,
+            msg.sid,
+            msg.reply_to orelse "",
+            msg.payload.len,
+            msg.payload,
+        },
+    );
+    try out.flush();
+}