From d0404059697d24d108f5759b164cbf3a85d92269 Mon Sep 17 00:00:00 2001 From: Karl Seguin Date: Tue, 4 Feb 2025 18:51:21 +0800 Subject: [PATCH] Remove TCP Mode / Server Websocket handler now interacts directly with cdp. --- .gitmodules | 2 +- src/browser/loader.zig | 3 +- src/cdp/browser.zig | 5 + src/cdp/cdp.zig | 1 + src/cdp/runtime.zig | 4 +- src/handler.zig | 95 ------- src/main.zig | 64 ++--- src/msg.zig | 140 ----------- src/server.zig | 548 +++++++---------------------------------- vendor/websocket.zig | 2 +- 10 files changed, 122 insertions(+), 742 deletions(-) delete mode 100644 src/handler.zig diff --git a/.gitmodules b/.gitmodules index 184dd2025..8ca411e81 100644 --- a/.gitmodules +++ b/.gitmodules @@ -30,5 +30,5 @@ url = https://github.com/lightpanda-io/zig-async-io.git/ [submodule "vendor/websocket.zig"] path = vendor/websocket.zig - url = https://github.com/lightpanda-io/websocket.zig.git/ + url = https://github.com/karlseguin/websocket.zig.git/ branch = lightpanda diff --git a/src/browser/loader.zig b/src/browser/loader.zig index 480ec4c1d..eb006a323 100644 --- a/src/browser/loader.zig +++ b/src/browser/loader.zig @@ -84,7 +84,8 @@ test "basic url get" { var loader = Loader.init(alloc); defer loader.deinit(); - var result = try loader.get(alloc, "https://en.wikipedia.org/wiki/Main_Page"); + const uri = try std.Uri.parse("https://en.wikipedia.org/wiki/Main_Page"); + var result = try loader.get(alloc, uri); defer result.deinit(); try std.testing.expect(result.req.response.status == std.http.Status.ok); diff --git a/src/cdp/browser.zig b/src/cdp/browser.zig index 12f882e2c..a853bfee0 100644 --- a/src/cdp/browser.zig +++ b/src/cdp/browser.zig @@ -32,6 +32,7 @@ const Methods = enum { setDownloadBehavior, getWindowForTarget, setWindowBounds, + close, }; pub fn browser( @@ -47,6 +48,10 @@ pub fn browser( .setDownloadBehavior => setDownloadBehavior(alloc, msg, ctx), .getWindowForTarget => getWindowForTarget(alloc, msg, ctx), .setWindowBounds => setWindowBounds(alloc, msg, ctx), + .close => { + ctx.state.close = true; + return ""; + } }; } diff --git a/src/cdp/cdp.zig b/src/cdp/cdp.zig index 65e2cc85b..a9e6b7123 100644 --- a/src/cdp/cdp.zig +++ b/src/cdp/cdp.zig @@ -119,6 +119,7 @@ pub fn dispatch( } pub const State = struct { + close: bool = false, executionContextId: u32 = 0, contextID: ?[]const u8 = null, sessionID: SessionID = .CONTEXTSESSIONID0497A05C95417CF4, diff --git a/src/cdp/runtime.zig b/src/cdp/runtime.zig index 44c1a9074..054d5a786 100644 --- a/src/cdp/runtime.zig +++ b/src/cdp/runtime.zig @@ -131,12 +131,12 @@ fn sendInspector( const buf = try alloc.alloc(u8, msg.json.len + 1); defer alloc.free(buf); _ = std.mem.replace(u8, msg.json, "\"awaitPromise\":true", "\"awaitPromise\":false", buf); - ctx.sendInspector(buf); + try ctx.sendInspector(buf); return ""; } } - ctx.sendInspector(msg.json); + try ctx.sendInspector(msg.json); if (msg.id == null) return ""; diff --git a/src/handler.zig b/src/handler.zig deleted file mode 100644 index 0decb3f78..000000000 --- a/src/handler.zig +++ /dev/null @@ -1,95 +0,0 @@ -// Copyright (C) 2023-2024 Lightpanda (Selecy SAS) -// -// Francis Bouvier -// Pierre Tachoire -// -// 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 . - -const std = @import("std"); - -const ws = @import("websocket"); -const Msg = @import("msg.zig").Msg; - -const log = std.log.scoped(.handler); - -pub const Stream = struct { - addr: std.net.Address, - socket: std.posix.socket_t = undefined, - - ws_host: []const u8, - ws_port: u16, - 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, 0, 0, 0 } ++ "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, data: []const u8) !void { - var header: [4]u8 = undefined; - Msg.setSize(data.len, &header); - try self.stream.recv(&header); - try self.stream.recv(data); - } -}; diff --git a/src/main.zig b/src/main.zig index e4da1df2f..44aa0e1b4 100644 --- a/src/main.zig +++ b/src/main.zig @@ -24,7 +24,6 @@ const websocket = @import("websocket"); const Browser = @import("browser/browser.zig").Browser; const server = @import("server.zig"); -const handler = @import("handler.zig"); const MaxSize = @import("msg.zig").MaxSize; const parser = @import("netsurf"); @@ -86,11 +85,9 @@ const CliMode = union(CliModeTag) { const Server = struct { execname: []const u8 = undefined, args: *std.process.ArgIterator = undefined, - addr: std.net.Address = undefined, host: []const u8 = Host, port: u16 = Port, timeout: u8 = Timeout, - tcp: bool = false, // undocumented TCP mode // default options const Host = "127.0.0.1"; @@ -160,10 +157,6 @@ 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, "--")) { @@ -186,10 +179,6 @@ const CliMode = union(CliModeTag) { if (default_mode == .server) { // server mode - _server.addr = std.net.Address.parseIp4(_server.host, _server.port) catch |err| { - log.err("address (host:port) {any}\n", .{err}); - return printUsageExit(execname, 1); - }; _server.execname = execname; _server.args = args; return CliMode{ .server = _server }; @@ -249,50 +238,19 @@ pub fn main() !void { .server => |opts| { // Stream server - const addr = blk: { - if (opts.tcp) { - break :blk opts.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(opts.execname, 1); - }; - defer std.posix.close(socket); - log.debug("Server opts: listening internally on {any}...", .{addr}); - - const timeout = std.time.ns_per_s * @as(u64, opts.timeout); // loop var loop = try jsruntime.Loop.init(alloc); defer loop.deinit(); - // TCP server mode - if (opts.tcp) { - return server.handle(alloc, &loop, socket, null, timeout); - } - - // start stream server in separate thread - var stream = handler.Stream{ - .ws_host = opts.host, - .ws_port = opts.port, - .addr = addr, - }; - const cdp_thread = try std.Thread.spawn( - .{ .allocator = alloc }, - server.handle, - .{ alloc, &loop, socket, &stream, timeout }, - ); + const vm = jsruntime.VM.init(); + defer vm.deinit(); // Websocket server - var ws = try websocket.Server(handler.Handler).init(alloc, .{ + var ws = try websocket.Server(server.Client).init(alloc, .{ .port = opts.port, .address = opts.host, - .max_message_size = MaxSize + 14, // overhead websocket + .max_message_size = 256 * 1024 + 14, // + 14 is the max websocket header len .max_conn = 1, .handshake = .{ .timeout = 3, @@ -304,8 +262,18 @@ pub fn main() !void { }); defer ws.deinit(); - try ws.listen(&stream); - cdp_thread.join(); + + try ws.listen(.{ + .vm = vm, + .loop = &loop, + .allocator = alloc, + // The websocket.zig fork has a hard-coded hack for handling + // puppeteer's request to /json/version. The hack relies on + // having these 2 fields to generating a response. These should + // be removed when the hack is removed. + .ws_host = opts.host, + .ws_port = opts.port, + }); }, .fetch => |opts| { diff --git a/src/msg.zig b/src/msg.zig index 13b7a62e9..ca6a28228 100644 --- a/src/msg.zig +++ b/src/msg.zig @@ -24,143 +24,3 @@ pub const MsgSize = 256 * 1204; // 256KB // but we prefer to put a lower hard limit for obvious memory size reasons. pub const MaxSize = HeaderSize + MsgSize; - -pub const Msg = struct { - pub fn getSize(data: []const u8) usize { - return std.mem.readInt(u32, data[0..HeaderSize], .little); - } - - pub fn setSize(len: usize, header: *[4]u8) void { - std.mem.writeInt(u32, header, @intCast(len), .little); - } -}; - -/// Buffer returns messages from a raw text read stream, -/// with the message size being encoded on the 2 first bytes (little endian) -/// It handles both: -/// - combined messages in one read -/// - single message in several reads (multipart) -/// It's safe (and a good practice) to reuse the same Buffer -/// on several reads of the same stream. -pub const Buffer = struct { - buf: []u8, - size: usize = 0, - pos: usize = 0, - - fn isFinished(self: *const Buffer) bool { - return self.pos >= self.size; - } - - fn isEmpty(self: *const Buffer) bool { - return self.size == 0 and self.pos == 0; - } - - fn reset(self: *Buffer) void { - self.size = 0; - self.pos = 0; - } - - // read input - pub fn read(self: *Buffer, input: []const u8) !struct { - msg: []const u8, - left: []const u8, - } { - var _input = input; // make input writable - - // msg size - var msg_size: usize = undefined; - if (self.isEmpty()) { - // decode msg size header - msg_size = Msg.getSize(_input); - _input = _input[HeaderSize..]; - } else { - msg_size = self.size; - } - - // multipart - const is_multipart = !self.isEmpty() or _input.len < msg_size; - if (is_multipart) { - - // set msg size on empty Buffer - if (self.isEmpty()) { - self.size = msg_size; - } - - // get the new position of the cursor - const new_pos = self.pos + _input.len; - - // check max limit size - if (new_pos > MaxSize) { - return error.MsgTooBig; - } - - // copy the current input into Buffer - // NOTE: we could use @memcpy but it's not Thread-safe (alias problem) - // see https://www.openmymind.net/Zigs-memcpy-copyForwards-and-copyBackwards/ - // Intead we just use std.mem.copyForwards - std.mem.copyForwards(u8, self.buf[self.pos..new_pos], _input[0..]); - - // set the new cursor position - self.pos = new_pos; - - // if multipart is not finished, go fetch the next input - if (!self.isFinished()) return error.MsgMultipart; - - // otherwhise multipart is finished, use its buffer as input - _input = self.buf[0..self.pos]; - self.reset(); - } - - // handle several JSON msg in 1 read - return .{ .msg = _input[0..msg_size], .left = _input[msg_size..] }; - } -}; - -test "Buffer" { - const Case = struct { - input: []const u8, - nb: u8, - }; - - const cases = [_]Case{ - // simple - .{ .input = .{ 2, 0, 0, 0 } ++ "ok", .nb = 1 }, - // combined - .{ .input = .{ 2, 0, 0, 0 } ++ "ok" ++ .{ 3, 0, 0, 0 } ++ "foo", .nb = 2 }, - // multipart - .{ .input = .{ 9, 0, 0, 0 } ++ "multi", .nb = 0 }, - .{ .input = "part", .nb = 1 }, - // multipart & combined - .{ .input = .{ 9, 0, 0, 0 } ++ "multi", .nb = 0 }, - .{ .input = "part" ++ .{ 2, 0, 0, 0 } ++ "ok", .nb = 2 }, - // multipart & combined with other multipart - .{ .input = .{ 9, 0, 0, 0 } ++ "multi", .nb = 0 }, - .{ .input = "part" ++ .{ 8, 0, 0, 0 } ++ "co", .nb = 1 }, - .{ .input = "mbined", .nb = 1 }, - // several multipart - .{ .input = .{ 23, 0, 0, 0 } ++ "multi", .nb = 0 }, - .{ .input = "several", .nb = 0 }, - .{ .input = "complex", .nb = 0 }, - .{ .input = "part", .nb = 1 }, - // combined & multipart - .{ .input = .{ 2, 0, 0, 0 } ++ "ok" ++ .{ 9, 0, 0, 0 } ++ "multi", .nb = 1 }, - .{ .input = "part", .nb = 1 }, - }; - - var b: [MaxSize]u8 = undefined; - var buf = Buffer{ .buf = &b }; - - for (cases) |case| { - var nb: u8 = 0; - var input = case.input; - while (input.len > 0) { - const parts = buf.read(input) catch |err| { - if (err == error.MsgMultipart) break; // go to the next case input - return err; - }; - nb += 1; - input = parts.left; - } - try std.testing.expect(nb == case.nb); - } -} diff --git a/src/server.zig b/src/server.zig index cbe16e129..275e46908 100644 --- a/src/server.zig +++ b/src/server.zig @@ -17,517 +17,157 @@ // along with this program. If not, see . const std = @import("std"); -const builtin = @import("builtin"); - -const Stream = @import("handler.zig").Stream; +const ws = @import("websocket"); +const cdp = @import("cdp/cdp.zig"); const jsruntime = @import("jsruntime"); -const Completion = jsruntime.IO.Completion; -const AcceptError = jsruntime.IO.AcceptError; -const RecvError = jsruntime.IO.RecvError; -const SendError = jsruntime.IO.SendError; -const CloseError = jsruntime.IO.CloseError; -const CancelError = jsruntime.IO.CancelError; -const TimeoutError = jsruntime.IO.TimeoutError; - -const MsgBuffer = @import("msg.zig").Buffer; -const MaxSize = @import("msg.zig").MaxSize; const Browser = @import("browser/browser.zig").Browser; -const cdp = @import("cdp/cdp.zig"); -const NoError = error{NoError}; -const IOError = AcceptError || RecvError || SendError || CloseError || TimeoutError || CancelError; -const Error = IOError || std.fmt.ParseIntError || cdp.Error || NoError; - -const TimeoutCheck = std.time.ns_per_ms * 100; +const Allocator = std.mem.Allocator; const log = std.log.scoped(.server); -const isLinux = builtin.target.os.tag == .linux; - -// I/O Main -// -------- -const BufReadSize = 1024; // 1KB -const MaxStdOutSize = 512; // ensure debug msg are not too long +// cdp works on a Ctx struct +pub const Ctx = Client; -pub const Ctx = struct { +pub const Client = struct { + browser: *Browser, + conn: *ws.Conn, loop: *jsruntime.Loop, - stream: ?*Stream, - - // internal fields - accept_socket: std.posix.socket_t, - conn_socket: std.posix.socket_t = undefined, - read_buf: []u8, // only for read operations - msg_buf: *MsgBuffer, - err: ?Error = null, - - // I/O fields - accept_completion: *Completion, - conn_completion: *Completion, - timeout_completion: *Completion, - timeout: u64, - last_active: ?std.time.Instant = null, - - // CDP - state: cdp.State = .{}, - - // JS fields - browser: *Browser, // TODO: is pointer mandatory here? - sessionNew: bool, - // try_catch: jsruntime.TryCatch, // TODO - - // callbacks - // --------- - - fn acceptCbk( - self: *Ctx, - completion: *Completion, - result: AcceptError!std.posix.socket_t, - ) void { - std.debug.assert(completion == self.acceptCompletion()); - - self.conn_socket = result catch |err| { - log.err("accept error: {any}", .{err}); - self.err = err; - return; - }; - log.info("client connected", .{}); - - // set connection timestamp and timeout - self.last_active = std.time.Instant.now() catch |err| { - log.err("accept timestamp error: {any}", .{err}); - return; - }; - self.loop.io.timeout( - *Ctx, - self, - Ctx.timeoutCbk, - self.timeout_completion, - TimeoutCheck, - ); - - // receving incomming messages asynchronously - self.loop.io.recv( - *Ctx, - self, - Ctx.readCbk, - self.conn_completion, - self.conn_socket, - self.read_buf, - ); - } - - fn readCbk(self: *Ctx, completion: *Completion, result: RecvError!usize) void { - std.debug.assert(completion == self.conn_completion); - - const size = result catch |err| { - if (self.isClosed() and err == error.FileDescriptorInvalid) { - log.debug("read has been canceled", .{}); - return; - } - log.err("read error: {any}", .{err}); - self.err = err; - return; - }; - - if (size == 0) { - // continue receving incomming messages asynchronously - self.loop.io.recv( - *Ctx, - self, - Ctx.readCbk, - self.conn_completion, - self.conn_socket, - self.read_buf, - ); - return; - } - - // set connection timestamp - self.last_active = std.time.Instant.now() catch |err| { - log.err("read timestamp error: {any}", .{err}); - return; - }; + allocator: Allocator, - // continue receving incomming messages asynchronously - self.loop.io.recv( - *Ctx, - self, - Ctx.readCbk, - self.conn_completion, - self.conn_socket, - self.read_buf, - ); + // used when gluing the session id to the inspector message + scrap: std.ArrayListUnmanaged(u8), - // read and execute input - var input: []const u8 = self.read_buf[0..size]; - while (input.len > 0) { - const parts = self.msg_buf.read(input) catch |err| { - if (err == error.MsgMultipart) { - return; - } else { - log.err("msg read error: {any}", .{err}); - return; - } - }; - input = parts.left; - // execute - self.do(parts.msg) catch |err| { - if (err != error.Closed) { - log.err("do error: {any}", .{err}); - log.debug("last msg: {s}", .{parts.msg}); - } - }; - } - } + // mutated by cdp directly + state: cdp.State, - fn timeoutCbk(self: *Ctx, completion: *Completion, result: TimeoutError!void) void { - std.debug.assert(completion == self.timeout_completion); + pub fn init(_: ws.Handshake, conn: *ws.Conn, context: anytype) !Client { + const loop = context.loop; + const allocator = context.allocator; - _ = result catch |err| { - log.err("timeout error: {any}", .{err}); - self.err = err; - return; - }; + const browser = try allocator.create(Browser); + errdefer allocator.destroy(browser); - if (self.isClosed()) { - // conn is already closed, ignore timeout - return; - } + try Browser.init(browser, allocator, loop, context.vm); + errdefer browser.deinit(); - // check time since last read - const now = std.time.Instant.now() catch |err| { - log.err("timeout timestamp error: {any}", .{err}); - return; + return .{ + .loop = loop, + .conn = conn, + .scrap = .{}, + .state = .{}, + .browser = browser, + .allocator = allocator, }; - - if (now.since(self.last_active.?) > self.timeout) { - // close current connection - log.debug("conn timeout, closing...", .{}); - self.close(); - return; - } - - // continue checking timeout - self.loop.io.timeout( - *Ctx, - self, - Ctx.timeoutCbk, - self.timeout_completion, - TimeoutCheck, - ); } - // shortcuts - // --------- - - inline fn isClosed(self: *Ctx) bool { - // last_active is first saved on acceptCbk - return self.last_active == null; + pub fn afterInit(self: *Client) !void { + try self.setupInspector(); } - // allocator of the current session - inline fn alloc(self: *Ctx) std.mem.Allocator { - return self.browser.session.alloc; + pub fn close(self: *Client) void { + self.browser.deinit(); } - // JS env of the current session - inline fn env(self: Ctx) jsruntime.Env { - return self.browser.session.env; + fn setupInspector(self: *Client) !void { + try self.browser.session.initInspector(self, inspectorResponse, inspectorEvent); } - inline fn acceptCompletion(self: *Ctx) *Completion { - // NOTE: the logical completion to use here is the accept_completion - // as the pipe_connection can be used simulteanously by a recv I/O operation. - // But on MacOS (kqueue) the recv I/O operation on a closed socket leads to a panic - // so we use the pipe_connection to avoid this problem - if (isLinux) return self.accept_completion; - return self.conn_completion; - } - - // actions - // ------- - - fn do(self: *Ctx, cmd: []const u8) anyerror!void { - - // close cmd - if (std.mem.eql(u8, cmd, "close")) { - // close connection - log.info("close cmd, closing conn...", .{}); - self.close(); - return error.Closed; - } - - if (self.sessionNew) self.sessionNew = false; - - const res = cdp.do(self.alloc(), cmd, self) catch |err| { - - // cdp end cmd + pub fn clientMessage(self: *Client, cmd: []const u8) !void { + const res = cdp.do(self.allocator, cmd, self) catch |err| { if (err == error.DisposeBrowserContext) { - // restart a new browser session std.log.scoped(.cdp).debug("end cmd, restarting a new session...", .{}); - try self.newSession(); + try self.browser.newSession(self.allocator, self.loop); + try self.setupInspector(); return; } - + log.err("CDP error: {}\nFor message: {s}", .{err, cmd}); return err; }; - // send result - if (!std.mem.eql(u8, res, "")) { - return self.send(res); - } - } - - pub fn send(self: *Ctx, msg: []const u8) !void { - if (self.stream) |stream| { - // if we have a stream connection, just write on it - defer self.alloc().free(msg); - try stream.send(msg); - } else { - // otherwise write asynchronously on the socket connection - return sendAsync(self, msg); - } - } - - fn close(self: *Ctx) void { - - // conn is closed - self.last_active = null; - std.posix.close(self.conn_socket); - log.debug("connection closed", .{}); - - // restart a new browser session in case of re-connect - if (!self.sessionNew) { - self.newSession() catch |err| { - log.err("new session error: {any}", .{err}); - return; - }; + if (res.len == 0) { + if (self.state.close) { + try self.conn.close(.{}); + } + return; } - log.info("accepting new conn...", .{}); - - // continue accepting incoming requests - self.loop.io.accept( - *Ctx, - self, - Ctx.acceptCbk, - self.acceptCompletion(), - self.accept_socket, - ); - } - - fn newSession(self: *Ctx) !void { - try self.browser.newSession(self.alloc(), self.loop); - try self.browser.session.initInspector( - self, - Ctx.onInspectorResp, - Ctx.onInspectorNotif, - ); - self.sessionNew = true; + return self.conn.write(res); } - // inspector - // --------- - - pub fn sendInspector(self: *Ctx, msg: []const u8) void { - if (self.env().getInspector()) |inspector| { - inspector.send(self.env(), msg); - } else @panic("Inspector has not been set"); - } - - inline fn inspectorCtx(ctx_opaque: *anyopaque) *Ctx { - const aligned = @as(*align(@alignOf(Ctx)) anyopaque, @alignCast(ctx_opaque)); - return @as(*Ctx, @ptrCast(aligned)); + // called by cdp + pub fn send(self: *Client, data: []const u8) !void { + return self.conn.write(data); } - fn inspectorMsg(allocator: std.mem.Allocator, ctx: *Ctx, msg: []const u8) !void { - // inject sessionID in cdp msg - const tpl = "{s},\"sessionId\":\"{s}\"}}"; - const msg_open = msg[0 .. msg.len - 1]; // remove closing bracket - const s = try std.fmt.allocPrint( - allocator, - tpl, - .{ msg_open, @tagName(ctx.state.sessionID) }, - ); - - try ctx.send(s); + // called by cdp + pub fn sendInspector(self: *Client, msg: []const u8) !void { + const env = self.browser.session.env; + if (env.getInspector()) |inspector| { + inspector.send(env, msg); + return; + } + return error.InspectNotSet; } - pub fn onInspectorResp(ctx_opaque: *anyopaque, _: u32, msg: []const u8) void { + pub fn inspectorResponse(ctx: *anyopaque, _: u32, msg: []const u8) void { if (std.log.defaultLogEnabled(.debug)) { // msg should be {"id":,... - const id_end = std.mem.indexOfScalar(u8, msg, ',') orelse unreachable; + std.debug.assert(std.mem.startsWith(u8, msg, "{\"id\":")); + + const id_end = std.mem.indexOfScalar(u8, msg, ',') orelse { + log.warn("invalid inspector response message: {s}", .{msg}); + return; + }; + const id = msg[6..id_end]; std.log.scoped(.cdp).debug("Res (inspector) > id {s}", .{id}); } - const ctx = inspectorCtx(ctx_opaque); - inspectorMsg(ctx.alloc(), ctx, msg) catch unreachable; + sendInspectorMessage(@alignCast(@ptrCast(ctx)), msg); } - pub fn onInspectorNotif(ctx_opaque: *anyopaque, msg: []const u8) void { + pub fn inspectorEvent(ctx: *anyopaque, msg: []const u8) void { if (std.log.defaultLogEnabled(.debug)) { // msg should be {"method":,... - const method_end = std.mem.indexOfScalar(u8, msg, ',') orelse unreachable; + std.debug.assert(std.mem.startsWith(u8, msg, "{\"method\":")); + const method_end = std.mem.indexOfScalar(u8, msg, ',') orelse { + log.warn("invalid inspector event message: {s}", .{msg}); + return; + }; const method = msg[10..method_end]; std.log.scoped(.cdp).debug("Event (inspector) > method {s}", .{method}); } - const ctx = inspectorCtx(ctx_opaque); - inspectorMsg(ctx.alloc(), ctx, msg) catch unreachable; + + sendInspectorMessage(@alignCast(@ptrCast(ctx)), msg); } -}; -// I/O Send -// -------- + fn sendInspectorMessage(self: *Client, msg: []const u8) void { + var scrap = &self.scrap; + scrap.clearRetainingCapacity(); -// NOTE: to allow concurrent send we create each time a dedicated context -// (with its own completion), allocated on the heap. -// After the send (on the sendCbk) the dedicated context will be destroy -// and the msg slice will be free. -const Send = struct { - ctx: *Ctx, - msg: []const u8, - completion: Completion = undefined, + const field = ",\"sessionId\":"; + const sessionID = @tagName(self.state.sessionID); - fn init(ctx: *Ctx, msg: []const u8) !*Send { - const sd = try ctx.alloc().create(Send); - sd.* = .{ .ctx = ctx, .msg = msg }; - return sd; - } + // + 2 for the quotes around the session + const message_len = msg.len + sessionID.len + 2 + field.len; - fn deinit(self: *Send) void { - self.ctx.alloc().free(self.msg); - self.ctx.alloc().destroy(self); - } + scrap.ensureTotalCapacity(self.allocator, message_len) catch |err| { + log.err("Failed to expand inspector buffer: {}", .{err}); + @panic("OOM"); + }; - fn asyncCbk(self: *Send, _: *Completion, result: SendError!usize) void { - _ = result catch |err| { - log.err("send error: {any}", .{err}); - self.ctx.err = err; + // -1 because we dont' want the closing brace '}' + scrap.appendSliceAssumeCapacity(msg[0..msg.len - 1]); + scrap.appendSliceAssumeCapacity(field); + scrap.appendAssumeCapacity('"'); + scrap.appendSliceAssumeCapacity(sessionID); + scrap.appendSliceAssumeCapacity("\"}"); + std.debug.assert(scrap.items.len == message_len); + + self.conn.write(scrap.items) catch |err| { + log.debug("Failed to write inspector message to client: {}", .{err}); + self.conn.close(.{}) catch {}; }; - self.deinit(); } }; - -pub fn sendAsync(ctx: *Ctx, msg: []const u8) !void { - const sd = try Send.init(ctx, msg); - ctx.loop.io.send(*Send, sd, Send.asyncCbk, &sd.completion, ctx.conn_socket, sd.msg); -} - -// Listener and handler -// -------------------- - -pub fn handle( - alloc: std.mem.Allocator, - loop: *jsruntime.Loop, - server_socket: std.posix.socket_t, - stream: ?*Stream, - timeout: u64, -) anyerror!void { - - // create v8 vm - const vm = jsruntime.VM.init(); - defer vm.deinit(); - - // browser - var browser: Browser = undefined; - try Browser.init(&browser, alloc, loop, vm); - defer browser.deinit(); - - // create buffers - var read_buf: [BufReadSize]u8 = undefined; - var buf: [MaxSize]u8 = undefined; - var msg_buf = MsgBuffer{ .buf = &buf }; - - // create I/O completions - var accept_completion: Completion = undefined; - var conn_completion: Completion = undefined; - var timeout_completion: Completion = undefined; - - // create I/O contexts and callbacks - // for accepting connections and receving messages - var ctx = Ctx{ - .loop = loop, - .stream = stream, - .browser = &browser, - .sessionNew = true, - .read_buf = &read_buf, - .msg_buf = &msg_buf, - .accept_socket = server_socket, - .timeout = timeout, - .accept_completion = &accept_completion, - .conn_completion = &conn_completion, - .timeout_completion = &timeout_completion, - }; - try browser.session.initInspector( - &ctx, - Ctx.onInspectorResp, - Ctx.onInspectorNotif, - ); - - // accepting connection asynchronously on internal server - log.info("accepting new conn...", .{}); - loop.io.accept(*Ctx, &ctx, Ctx.acceptCbk, ctx.acceptCompletion(), ctx.accept_socket); - - // infinite loop on I/O events, either: - // - cmd from incoming connection on server socket - // - JS callbacks events from scripts - while (true) { - try loop.io.run_for_ns(10 * std.time.ns_per_ms); - if (loop.cbk_error) { - log.err("JS error", .{}); - // if (try try_catch.exception(alloc, js_env.*)) |msg| { - // std.debug.print("\n\rUncaught {s}\n\r", .{msg}); - // alloc.free(msg); - // } - // loop.cbk_error = false; - } - if (ctx.err) |err| { - if (err != error.NoError) log.err("Server error: {any}", .{err}); - break; - } - } -} - -fn setSockOpt(fd: std.posix.socket_t, level: i32, option: u32, value: c_int) !void { - 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 { - const isunixsock = isUnixSocket(address); - - // create socket - const flags = std.posix.SOCK.STREAM | std.posix.SOCK.CLOEXEC | std.posix.SOCK.NONBLOCK; - const proto = if (isunixsock) @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 - // - // REUSEPORT can't be set on unix socket anymore. - // see https://github.com/torvalds/linux/commit/5b0af621c3f6ef9261cf6067812f2fd9943acb4b - if (@hasDecl(std.posix.SO, "REUSEPORT") and !isunixsock) { - try setSockOpt(sockfd, std.posix.SOL.SOCKET, std.posix.SO.REUSEPORT, 1); - } - try setSockOpt(sockfd, std.posix.SOL.SOCKET, std.posix.SO.REUSEADDR, 1); - if (!isUnixSocket(address)) { - if (builtin.target.os.tag == .linux) { // posix.TCP not available on MacOS - // WARNING: disable Nagle's alogrithm to avoid latency issues - try setSockOpt(sockfd, std.posix.IPPROTO.TCP, std.posix.TCP.NODELAY, 1); - } - } - - // bind & listen - var socklen = address.getOsSockLen(); - try std.posix.bind(sockfd, &address.any, socklen); - const kernel_backlog = 1; // default value is 128. Here we just want 1 connection - try std.posix.listen(sockfd, kernel_backlog); - var listen_address: std.net.Address = undefined; - try std.posix.getsockname(sockfd, &listen_address.any, &socklen); - - return sockfd; -} diff --git a/vendor/websocket.zig b/vendor/websocket.zig index 1b49626c7..28c31c159 160000 --- a/vendor/websocket.zig +++ b/vendor/websocket.zig @@ -1 +1 @@ -Subproject commit 1b49626c78bb5063cede4f22b30496fc66e7ba64 +Subproject commit 28c31c1591660285b295bb87a599cbfbcb087819