websocket: use Unix socket for internal server

And add an option for TCP only server

Signed-off-by: Francis Bouvier <francis@lightpanda.io>

src/handler.zig

@@ -23,20 +23,22 @@ const ws = @import("websocket");
 const log = std.log.scoped(.handler);
 
 pub const Stream = struct {
+    addr: std.net.Address,
     socket: std.posix.socket_t = undefined,
     ws_conn: *ws.Conn = undefined,
 
     fn connectCDP(self: *Stream) !void {
-        const address = try std.net.Address.parseIp("127.0.0.1", 3245);
-
         const flags: u32 = std.posix.SOCK.STREAM;
-        const proto = std.posix.IPPROTO.TCP;
+        const proto = blk: {
-        const socket = try std.posix.socket(address.any.family, flags, proto);
+            if (self.addr.any.family == std.posix.AF.UNIX) break :blk @as(u32, 0);
+            break :blk std.posix.IPPROTO.TCP;
+        };
+        const socket = try std.posix.socket(self.addr.any.family, flags, proto);
 
         try std.posix.connect(
             socket,
-            &address.any,
+            &self.addr.any,
-            address.getOsSockLen(),
+            self.addr.getOsSockLen(),
         );
         log.debug("connected to Stream server", .{});
         self.socket = socket;

src/main.zig

@@ -80,10 +80,11 @@ const CliMode = union(CliModeTag) {
         host: []const u8 = Host,
         port: u16 = Port,
         timeout: u8 = Timeout,
+        tcp: bool = false, // undocumented TCP mode
 
         // default options
         const Host = "127.0.0.1";
-        const Port = 3245;
+        const Port = 9222;
         const Timeout = 3; // in seconds
     };
 
@@ -145,6 +146,10 @@ const CliMode = union(CliModeTag) {
                     return printUsageExit(execname, 1);
                 }
             }
+            if (std.mem.eql(u8, "--tcp", opt)) {
+                _server.tcp = true;
+                continue;
+            }
 
             // unknown option
             if (std.mem.startsWith(u8, opt, "--")) {
@@ -230,36 +235,45 @@ pub fn main() !void {
         .server => |mode| {
 
             // Stream server
-            const socket = server.listen(mode.addr) catch |err| {
+            const addr = blk: {
-                log.err("address (host:port) {any}\n", .{err});
+                if (mode.tcp) {
+                    break :blk mode.addr;
+                } else {
+                    const unix_path = "/tmp/lightpanda";
+                    std.fs.deleteFileAbsolute(unix_path) catch {}; // file could not exists
+                    break :blk try std.net.Address.initUnix(unix_path);
+                }
+            };
+            const socket = server.listen(addr) catch |err| {
+                log.err("Server listen error: {any}\n", .{err});
                 return printUsageExit(mode.execname, 1);
             };
             defer std.posix.close(socket);
-            log.debug("Server mode: listening internally on {s}:{d}...", .{ mode.host, mode.port });
+            log.debug("Server mode: listening internally on {any}...", .{addr});
 
-            var stream = handler.Stream{};
+            const timeout = std.time.ns_per_s * @as(u64, mode.timeout);
 
             // loop
             var loop = try jsruntime.Loop.init(alloc);
             defer loop.deinit();
 
+            // TCP server mode
+            if (mode.tcp) {
+                return server.handle(alloc, &loop, socket, null, timeout);
+            }
+
             // start stream server in separate thread
+            var stream = handler.Stream{ .addr = addr };
             const cdp_thread = try std.Thread.spawn(
                 .{ .allocator = alloc },
                 server.handle,
-                .{
+                .{ alloc, &loop, socket, &stream, timeout },
-                    alloc,
-                    &loop,
-                    socket,
-                    &stream,
-                    std.time.ns_per_s * @as(u64, mode.timeout),
-                },
             );
 
             // Websocket server
             var ws = try websocket.Server(handler.Handler).init(alloc, .{
-                .port = 9222,
+                .port = mode.port,
-                .address = "127.0.0.1",
+                .address = mode.host,
                 .handshake = .{
                     .timeout = 3,
                     .max_size = 1024,

src/server.zig

@@ -518,21 +518,29 @@ fn setSockOpt(fd: std.posix.socket_t, level: i32, option: u32, value: c_int) !vo
     try std.posix.setsockopt(fd, level, option, &std.mem.toBytes(value));
 }
 
+fn isUnixSocket(addr: std.net.Address) bool {
+    return addr.any.family == std.posix.AF.UNIX;
+}
+
 pub fn listen(address: std.net.Address) !std.posix.socket_t {
 
     // create socket
     const flags = std.posix.SOCK.STREAM | std.posix.SOCK.CLOEXEC | std.posix.SOCK.NONBLOCK;
-    const sockfd = try std.posix.socket(address.any.family, flags, std.posix.IPPROTO.TCP);
+    const proto = if (isUnixSocket(address)) @as(u32, 0) else std.posix.IPPROTO.TCP;
+    const sockfd = try std.posix.socket(address.any.family, flags, proto);
     errdefer std.posix.close(sockfd);
 
     // socket options
-    try setSockOpt(sockfd, std.posix.SOL.SOCKET, std.posix.SO.REUSEADDR, 1);
     if (@hasDecl(std.posix.SO, "REUSEPORT")) {
         try setSockOpt(sockfd, std.posix.SOL.SOCKET, std.posix.SO.REUSEPORT, 1);
+    } else {
+        try setSockOpt(sockfd, std.posix.SOL.SOCKET, std.posix.SO.REUSEADDR, 1);
     }
-    if (builtin.target.os.tag == .linux) { // posix.TCP not available on MacOS
+    if (!isUnixSocket(address)) {
-        // WARNING: disable Nagle's alogrithm to avoid latency issues
+        if (builtin.target.os.tag == .linux) { // posix.TCP not available on MacOS
-        try setSockOpt(sockfd, std.posix.IPPROTO.TCP, std.posix.TCP.NODELAY, 1);
+            // WARNING: disable Nagle's alogrithm to avoid latency issues
+            try setSockOpt(sockfd, std.posix.IPPROTO.TCP, std.posix.TCP.NODELAY, 1);
+        }
     }
 
     // bind & listen