zits/src/server/client.zig

const Message = @import("message_parser.zig").Message;
const std = @import("std");

const Client = @This();

connect: ?Message.Connect,

// Messages for this client to receive.
recv_queue: ?*std.Io.Queue(Message) = null,

from_client: *std.Io.Reader,
to_client: *std.Io.Writer,

pub fn init(
    connect: ?Message.Connect,
    in: *std.Io.Reader,
    out: *std.Io.Writer,
) Client {
    return .{
        .connect = connect,
        .from_client = in,
        .to_client = out,
    };
}

pub fn start(self: *Client, io: std.Io, alloc: std.mem.Allocator, queue: *std.Io.Queue(Message)) !void {
    self.recv_queue = queue;
    var msgs: [8]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});
                },
            }
        }
        try self.to_client.flush();
    }
}

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", .{});
    // defer std.debug.print("client returning next message!\n", .{});
    return Message.next(allocator, self.from_client);
    // return self.send_queue.getOne(io);
}

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

fn writeErr(out: *std.Io.Writer, msg: []const u8) !void {
    _ = try out.print("-ERR '{s}'\r\n", .{msg});
}

fn writePong(out: *std.Io.Writer) !void {
    _ = try out.write("PONG\r\n");
}

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

fn writeMsg(out: *std.Io.Writer, msg: Message.Msg) !void {
    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,
        },
    );
}

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

    {
        // Simulate stream
        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 |err| switch (err) {
            error.EndOfStream => try from_client_queue.close(io),
            else => return err,
        }

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