websocket: first implementation

.gitmodules

@@ -28,3 +28,7 @@
 [submodule "vendor/zig-async-io"]
 	path = vendor/zig-async-io
 	url = git@github.com:lightpanda-io/zig-async-io.git
+[submodule "vendor/websocket.zig"]
+	path = vendor/websocket.zig
+	url = git@github.com:lightpanda-io/websocket.zig.git
+	branch = lightpanda

build.zig

@@ -168,6 +168,11 @@ fn common(
         .root_source_file = b.path("vendor/tls.zig/src/main.zig"),
     });
     step.root_module.addImport("tls", tlsmod);
+
+    const wsmod = b.addModule("websocket", .{
+        .root_source_file = b.path("vendor/websocket.zig/src/websocket.zig"),
+    });
+    step.root_module.addImport("websocket", wsmod);
 }
 
 fn moduleNetSurf(b: *std.Build, target: std.Build.ResolvedTarget) !*std.Build.Module {

src/cdp/cdp.zig

@@ -193,7 +193,7 @@ pub fn sendEvent(
     const resp = Resp{ .method = name, .params = params, .sessionId = sessionID };
 
     const event_msg = try stringify(alloc, resp);
-    try server.sendAsync(ctx, event_msg);
+    try ctx.send(event_msg);
 }
 
 // Common

src/cdp/page.zig

@@ -323,7 +323,7 @@ fn navigate(
         .loaderId = ctx.state.loaderID,
     };
     const res = try result(alloc, input.id, Resp, resp, input.sessionId);
-    try server.sendAsync(ctx, res);
+    try ctx.send(res);
 
     // TODO: at this point do we need async the following actions to be async?
 

src/cdp/target.zig

@@ -292,7 +292,7 @@ fn disposeBrowserContext(
 
     // output
     const res = try result(alloc, input.id, null, .{}, null);
-    try server.sendAsync(ctx, res);
+    try ctx.send(res);
 
     return error.DisposeBrowserContext;
 }
@@ -378,7 +378,7 @@ fn closeTarget(
         success: bool = true,
     };
     const res = try result(alloc, input.id, Resp, Resp{}, null);
-    try server.sendAsync(ctx, res);
+    try ctx.send(res);
 
     // Inspector.detached event
     const InspectorDetached = struct {

src/handler.zig

@@ -0,0 +1,89 @@
+// Copyright (C) 2023-2024  Lightpanda (Selecy SAS)
+//
+// Francis Bouvier <francis@lightpanda.io>
+// Pierre Tachoire <pierre@lightpanda.io>
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+const std = @import("std");
+
+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 flags: u32 = std.posix.SOCK.STREAM;
+        const proto = blk: {
+            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,
+            &self.addr.any,
+            self.addr.getOsSockLen(),
+        );
+        log.debug("connected to Stream server", .{});
+        self.socket = socket;
+    }
+
+    fn closeCDP(self: *const Stream) void {
+        const close_msg: []const u8 = "5:close";
+        self.recv(close_msg) catch |err| {
+            log.err("stream close error: {any}", .{err});
+        };
+        std.posix.close(self.socket);
+    }
+
+    fn start(self: *Stream, ws_conn: *ws.Conn) !void {
+        try self.connectCDP();
+        self.ws_conn = ws_conn;
+    }
+
+    pub fn recv(self: *const Stream, data: []const u8) !void {
+        var pos: usize = 0;
+        while (pos < data.len) {
+            const len = try std.posix.write(self.socket, data[pos..]);
+            pos += len;
+        }
+    }
+
+    pub fn send(self: *const Stream, data: []const u8) !void {
+        return self.ws_conn.write(data);
+    }
+};
+
+pub const Handler = struct {
+    stream: *Stream,
+
+    pub fn init(_: ws.Handshake, ws_conn: *ws.Conn, stream: *Stream) !Handler {
+        try stream.start(ws_conn);
+        return .{ .stream = stream };
+    }
+
+    pub fn close(self: *Handler) void {
+        self.stream.closeCDP();
+    }
+
+    pub fn clientMessage(self: *Handler, alloc: std.mem.Allocator, data: []const u8) !void {
+        const msg = try std.fmt.allocPrint(alloc, "{d}:{s}", .{ data.len, data });
+        try self.stream.recv(msg);
+    }
+};