diff options
Diffstat (limited to 'src/server/client.zig')
| -rw-r--r-- | src/server/client.zig | 147 |
1 files changed, 113 insertions, 34 deletions
diff --git a/src/server/client.zig b/src/server/client.zig index b4bc55b..f8d4c93 100644 --- a/src/server/client.zig +++ b/src/server/client.zig @@ -3,7 +3,7 @@ const std = @import("std"); pub const ClientState = struct { id: usize, - connect: Message.Connect, + connect: Message.AllocatedConnect, /// Messages that this client is trying to send. send_queue: std.Io.Queue(Message), send_queue_buffer: [1024]Message = undefined, @@ -11,17 +11,16 @@ pub const ClientState = struct { 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, - 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, + connect: Message.AllocatedConnect, in: *std.Io.Reader, out: *std.Io.Writer, ) !ClientState { @@ -32,15 +31,11 @@ pub const ClientState = struct { .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); - // 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 }); + try res.io_group.concurrent(io, processWrite, .{ &res, io, out }); + try res.io_group.concurrent(io, processRead, .{ &res, io, allocator, in }); return res; } @@ -78,24 +73,30 @@ pub const ClientState = struct { allocator: std.mem.Allocator, in: *std.Io.Reader, ) void { - io.sleep(.fromMilliseconds(100), .real) catch @panic("couldn't sleep"); + // io.sleep(.fromMilliseconds(100), .real) catch @panic("couldn't sleep"); while (true) { - std.debug.print("waiting for message\n", .{}); + // std.debug.print("waiting for message\n", .{}); const next_message = Message.next(allocator, in) catch |err| switch (err) { - error.EndOfStream => break, + error.EndOfStream => { + self.send_queue.putOne(io, .eos) catch {}; + break; + }, else => { std.debug.panic("guh: {any}\n", .{err}); break; - // return err; }, }; - 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("next_message: {any}\n", .{next_message}); - // std.debug.print("queue: {any}\n", .{self.send_queue}); - // }; + 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", .{}); @@ -171,21 +172,99 @@ test { 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 => {}, - } + { + 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); |