const std = @import("std"); const Message = @import("./message_parser.zig").Message; pub const ServerInfo = Message.ServerInfo; const ClientState = @import("./client.zig").ClientState; const Server = @This(); info: ServerInfo, clients: std.AutoHashMapUnmanaged(usize, *ClientState) = .empty, /// Map of subjects to a map of (client ID => SID) subscriptions: std.StringHashMapUnmanaged(std.AutoHashMapUnmanaged(usize, []const u8)) = .empty, var keep_running = std.atomic.Value(bool).init(true); fn handleSigInt(sig: std.os.linux.SIG) callconv(.c) void { _ = sig; keep_running.store(false, .monotonic); } pub fn main(gpa: std.mem.Allocator, server_config: ServerInfo) !void { // Configure the signal action // const act = std.posix.Sigaction{ // .handler = .{ .handler = handleSigInt }, // .mask = std.posix.sigemptyset(), // .flags = 0, // }; // // Register the handler for SIGINT (Ctrl+C) // std.posix.sigaction(std.posix.SIG.INT, &act, null); var server: Server = .{ .info = server_config, }; var threaded: std.Io.Threaded = .init(gpa); defer threaded.deinit(); const io = threaded.io(); var tcp_server = try std.Io.net.IpAddress.listen(try std.Io.net.IpAddress.parse( server.info.host, server.info.port, ), io, .{}); defer tcp_server.deinit(io); var id: usize = 0; // Run until SIGINT is handled, then exit gracefully while (keep_running.load(.monotonic)) : (id +%= 1) { std.debug.print("in server loop\n", .{}); if (server.clients.contains(id)) continue; const stream = try tcp_server.accept(io); std.debug.print("accepted connection\n", .{}); _ = io.concurrent(handleConnection, .{ &server, gpa, io, id, stream }) catch { std.debug.print("could not start concurrent handler for {d}\n", .{id}); stream.close(io); }; } std.debug.print("Exiting gracefully\n", .{}); } fn addClient(server: *Server, allocator: std.mem.Allocator, id: usize, client: *ClientState) !void { // server.clients.lockPointers(); try server.clients.put(allocator, id, client); // server.clients.unlockPointers(); } fn removeClient(server: *Server, allocator: std.mem.Allocator, id: usize) void { // TODO: implement _ = server; _ = allocator; _ = id; } fn handleConnection( server: *Server, server_allocator: std.mem.Allocator, io: std.Io, id: usize, stream: std.Io.net.Stream, ) !void { var client_allocator: std.heap.DebugAllocator(.{}) = .init; client_allocator.backing_allocator = server_allocator; defer { std.debug.print("deinitializing debug allocator\n", .{}); _ = client_allocator.deinit(); } const allocator = client_allocator.allocator(); defer stream.close(io); var w_buffer: [1024]u8 = undefined; var writer = stream.writer(io, &w_buffer); const out = &writer.interface; std.debug.print("out pointer in client handler: {*}\n", .{out}); var r_buffer: [8192]u8 = undefined; var reader = stream.reader(io, &r_buffer); const in = &reader.interface; @import("./client.zig").writeInfo(out, server.info) catch return; var connect_arena: std.heap.ArenaAllocator = .init(allocator); defer connect_arena.deinit(); const connect = (Message.next(connect_arena.allocator(), in) catch return).connect; var client_state: ClientState = try .init(io, allocator, id, connect, in, out); defer client_state.deinit(io, allocator); try server.addClient(allocator, id, &client_state); defer server.removeClient(allocator, id); while (client_state.next(allocator)) |msg| { std.debug.print("message from client: {any}\n", .{msg}); switch (msg) { .ping => { // Respond to ping with pong. for (0..5) |_| { if (try client_state.send(io, .pong)) { break; } } else {} }, .@"pub" => |@"pub"| { std.debug.print("pub: {any}\n", .{@"pub"}); try server.publishMessage(io, @"pub"); if (client_state.connect.connect.verbose) { std.debug.print("server IS sending +ok\n", .{}); _ = try client_state.send(io, .@"+ok"); } else { std.debug.print("server NOT sending +ok\n", .{}); } }, .sub => |sub| { try server.subscribe(allocator, client_state.id, sub); }, .unsub => |unsub| { try server.unsubscribe(client_state.id, unsub); }, else => |e| { std.debug.panic("Unimplemented message: {any}\n", .{e}); }, } std.debug.print("processed message from client\n", .{}); std.debug.print("awaiting next message from client\n", .{}); } else |err| { // This is probably going to be normal on disconnect std.debug.print("Ran into error in client process loop: {}\n", .{err}); } // client_state.task.await(io); } // // Result is owned by the caller // fn subscribers(server: *Server, gpa: std.mem.Allocator, subject: []const u8) []ClientState { // var acc: std.ArrayList(ClientState) = .empty; // return acc.toOwnedSlice(); // } fn publishMessage(server: *Server, io: std.Io, msg: Message.Pub) !void { if (server.subscriptions.get(msg.subject)) |subs| { var subs_iter = subs.iterator(); while (subs_iter.next()) |sub| { const client_id = sub.key_ptr.*; const sid = sub.value_ptr.*; const client = server.clients.getPtr(client_id) orelse { std.debug.print("trying to publish to a client that no longer exists: {d}", .{client_id}); continue; }; _ = try client.*.send(io, .{ .msg = .{ .subject = msg.subject, .sid = sid, .reply_to = msg.reply_to, .payload = msg.payload, } }); } } else { std.debug.print("no subs on {s}\n", .{msg.subject}); } } fn subscribe(server: *Server, gpa: std.mem.Allocator, id: usize, msg: Message.Sub) !void { var subs_for_subject: std.AutoHashMapUnmanaged(usize, []const u8) = if (server.subscriptions.fetchRemove(msg.subject)) |s| s.value else .empty; try subs_for_subject.put(gpa, id, msg.sid); try server.subscriptions.put(gpa, msg.subject, subs_for_subject); } fn unsubscribe(server: *Server, id: usize, msg: Message.Unsub) !void { // Get the subscription in subscriptions by looping over all the subjects, // and getting the SID for that subject for the current client ID. // If the SID matches, remove the kv for the client ID from subscriptions for that subject. // If the value for that subject is empty, remove the subject. var subscriptions_iter = server.subscriptions.iterator(); while (subscriptions_iter.next()) |*subs_for_sub| { if (subs_for_sub.value_ptr.get(id)) |client_sub| { if (std.mem.eql(u8, client_sub, msg.sid)) { _ = subs_for_sub.value_ptr.*.remove(id); if (subs_for_sub.value_ptr.count() == 0) { _ = server.subscriptions.remove(subs_for_sub.key_ptr.*); } break; } } } } pub fn createId() []const u8 { return "SERVERID"; } pub fn createName() []const u8 { return "SERVERNAME"; } // TESTING // fn initTestServer() Server { // return .{ // .info = .{ // .server_id = "ABCD", // .server_name = "test server", // .version = "0.1.2", // .max_payload = 1234, // }, // }; // } // fn initTestClient( // io: std.Io, // allocator: std.mem.Allocator, // id: usize, // data_from: []const u8, // ) !struct { // Client, // *std.Io.Reader, // *std.Io.Writer, // } { // return .init(io, allocator, id, .{}, in, out); // } // test { // const gpa = std.testing.allocator; // const io = std.testing.io; // const server = initTestServer(); // const client: Client = .init( // io, // gpa, // 1, // .{}, // ); // }