zits/src/server/client.zig

const Message = @import("message_parser.zig").Message;
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,

    pub fn init(
        io: std.Io,
        allocator: std.mem.Allocator,
        id: usize,
        connect: Message.AllocatedConnect,
        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,
        };
        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 });
        try res.io_group.concurrent(io, processRead, .{ &res, io, allocator, in });

        return res;
    }

    fn processWrite(
        self: *ClientState,
        io: std.Io,
        out: *std.Io.Writer,
    ) void {
        while (true) {
            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", .{});
                },
            }
        }
    }

    fn processRead(
        self: *ClientState,
        io: std.Io,
        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;
    }

    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);
    }
};

fn writeOk(out: *std.Io.Writer) !void {
    _ = try out.write("+OK\r\n");
    try out.flush();
}

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");
    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();
}

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| {
            switch (msg) {
                .eos => {
                    try from_client_queue.putOne(io, msg);
                    break;
                },
                else => {
                    try from_client_queue.putOne(io, msg);
                },
            }
        } else |_| {}

        while (from_client_queue.getOne(io)) |msg| {
            switch (msg) {
                .eos => {
                    break;
                },
                .connect => |*c| {
                    std.debug.print("Message: {any}\n", .{msg});
                    c.deinit();
                },
                else => {
                    std.debug.print("Message: {any}\n", .{msg});
                },
            }
        } else |_| {}
    }

    from_client_queue = .init(&from_client_buf);
    // Reset the reader to process it again.
    from_client.seek = 0;

    {
        const SemiClient = struct {
            q: std.Io.Queue(Message),

            fn parseClientInput(self: *@This(), ioh: std.Io, in: *std.Io.Reader) void {
                defer std.debug.print("done parse\n", .{});
                while (Message.next(gpa, in)) |msg| {
                    switch (msg) {
                        .eos => {
                            self.q.putOne(ioh, msg) catch return;
                            break;
                        },
                        else => {
                            self.q.putOne(ioh, msg) catch return;
                        },
                    }
                } else |_| {}
            }

            fn next(self: *@This(), ioh: std.Io) !Message {
                return self.q.getOne(ioh);
            }

            fn printAll(self: *@This(), ioh: std.Io) void {
                defer std.debug.print("done print\n", .{});
                while (self.next(ioh)) |*msg| {
                    switch (msg.*) {
                        .eos => |_| {
                            break;
                        },
                        else => {},
                    }
                    std.debug.print("Client msg: {any}\n", .{msg});
                    switch (msg.*) {
                        .connect => |c| {
                            // c.allocator.deinit();
                            c.deinit();
                            // @constCast(c).deinit();
                        },
                        else => {},
                    }
                } else |_| {}
            }
        };

        var c: SemiClient = .{ .q = from_client_queue };
        var group: std.Io.Group = .init;
        defer group.wait(io);

        group.concurrent(io, SemiClient.printAll, .{ &c, io }) catch {
            @panic("could not start printAll\n");
        };

        group.concurrent(io, SemiClient.parseClientInput, .{ &c, io, &from_client }) catch {
            @panic("could not start printAll\n");
        };
    }

    ////////

    // 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;
    //         },
    //     }
    // }
}