public/browser
1
0
Fork 0
mirror of https://github.com/lightpanda-io/browser.git synced 2025-10-29 15:13:28 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

async: adapt async cli

Browse Source
This commit is contained in:
Pierre Tachoire
2024-11-15 14:37:19 +01:00
parent fadf3f609a
commit 6809bb5393
7 changed files with 282 additions and 295 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/browser/browser.zig
View File

@@ -40,7 +40,7 @@ const storage = @import("../storage/storage.zig");
const FetchResult = @import("../http/Client.zig").Client.FetchResult;
const UserContext = @import("../user_context.zig").UserContext;
const HttpClient = @import("../async/Client.zig");
const HttpClient = @import("../http/async/main.zig").Client;
const log = std.log.scoped(.browser);
@@ -116,7 +116,7 @@ pub const Session = struct {
};
Env.init(&self.env, self.arena.allocator(), loop, null);
self.httpClient = .{ .allocator = alloc, .loop = loop };
self.httpClient = .{ .allocator = alloc };
try self.env.load(&self.jstypes);
}

71
src/http/async/io.zig
View File

@@ -1,6 +1,8 @@
const std = @import("std");
pub const IO = @import("jsruntime").IO;
const Ctx = @import("std/http/Client.zig").Ctx;
const Loop = @import("jsruntime").Loop;
const NetworkImpl = Loop.Network(SingleThreaded);
pub const Blocking = struct {
pub fn connect(
@@ -57,65 +59,55 @@ pub const Blocking = struct {
}
};
pub fn SingleThreaded(comptime CtxT: type) type {
return struct {
io: *IO,
completion: IO.Completion,
ctx: *CtxT,
cbk: CbkT,
count: u32 = 0,
const CbkT = *const fn (ctx: *CtxT, res: anyerror!void) anyerror!void;
pub const SingleThreaded = struct {
impl: NetworkImpl,
cbk: Cbk,
ctx: *Ctx,
const Self = @This();
const Cbk = *const fn (ctx: *Ctx, res: anyerror!void) anyerror!void;
pub fn init(io: *IO) Self {
pub fn init(loop: *Loop) Self {
return .{
.io = io,
.completion = undefined,
.ctx = undefined,
.impl = NetworkImpl.init(loop),
.cbk = undefined,
.ctx = undefined,
};
}
pub fn connect(
self: *Self,
comptime _: type,
ctx: *CtxT,
comptime cbk: CbkT,
ctx: *Ctx,
comptime cbk: Cbk,
socket: std.posix.socket_t,
address: std.net.Address,
) void {
self.ctx = ctx;
self.cbk = cbk;
self.count += 1;
self.io.connect(*Self, self, Self.connectCbk, &self.completion, socket, address);
self.ctx = ctx;
self.impl.connect(self, socket, address);
}
fn connectCbk(self: *Self, _: *IO.Completion, result: IO.ConnectError!void) void {
defer self.count -= 1;
_ = result catch |e| return self.ctx.setErr(e);
pub fn onConnect(self: *Self, err: ?anyerror) void {
if (err) |e| return self.ctx.setErr(e);
self.cbk(self.ctx, {}) catch |e| self.ctx.setErr(e);
}
pub fn send(
self: *Self,
comptime _: type,
ctx: *CtxT,
comptime cbk: CbkT,
ctx: *Ctx,
comptime cbk: Cbk,
socket: std.posix.socket_t,
buf: []const u8,
) void {
self.ctx = ctx;
self.cbk = cbk;
self.count += 1;
self.io.send(*Self, self, Self.sendCbk, &self.completion, socket, buf);
self.impl.send(self, socket, buf);
}
fn sendCbk(self: *Self, _: *IO.Completion, result: IO.SendError!usize) void {
defer self.count -= 1;
const ln = result catch |e| return self.ctx.setErr(e);
pub fn onSend(self: *Self, ln: usize, err: ?anyerror) void {
if (err) |e| return self.ctx.setErr(e);
self.ctx.setLen(ln);
self.cbk(self.ctx, {}) catch |e| self.ctx.setErr(e);
}
@@ -123,26 +115,19 @@ pub fn SingleThreaded(comptime CtxT: type) type {
pub fn recv(
self: *Self,
comptime _: type,
ctx: *CtxT,
comptime cbk: CbkT,
ctx: *Ctx,
comptime cbk: Cbk,
socket: std.posix.socket_t,
buf: []u8,
) void {
self.ctx = ctx;
self.cbk = cbk;
self.count += 1;
self.io.recv(*Self, self, Self.receiveCbk, &self.completion, socket, buf);
self.impl.receive(self, socket, buf);
}
fn receiveCbk(self: *Self, _: *IO.Completion, result: IO.RecvError!usize) void {
defer self.count -= 1;
const ln = result catch |e| return self.ctx.setErr(e);
pub fn onReceive(self: *Self, ln: usize, err: ?anyerror) void {
if (err) |e| return self.ctx.setErr(e);
self.ctx.setLen(ln);
self.cbk(self.ctx, {}) catch |e| self.ctx.setErr(e);
}
pub fn isDone(self: *Self) bool {
return self.count == 0;
}
};
}
};

99
src/http/async/std/http/Client.zig
View File

@@ -22,7 +22,7 @@ const tls23 = @import("../../tls.zig/main.zig");
const VecPut = @import("../../tls.zig/connection.zig").VecPut;
const GenericStack = @import("../../stack.zig").Stack;
const async_io = @import("../../io.zig");
pub const Loop = async_io.SingleThreaded(Ctx);
pub const Loop = async_io.SingleThreaded;
const cipher = @import("../../tls.zig/cipher.zig");
@@ -1343,8 +1343,29 @@ pub const Request = struct {
}
}
fn onWriteAll(ctx: *Ctx, res: anyerror!void) !void {
res catch |err| return ctx.pop(err);
switch (ctx.req.transfer_encoding) {
.chunked => unreachable,
.none => unreachable,
.content_length => |*len| {
len.* = 0;
},
}
try ctx.pop({});
}
pub fn async_writeAll(req: *Request, buf: []const u8, ctx: *Ctx, comptime cbk: Cbk) !void {
try req.connection.?.async_writeAllDirect(buf, ctx, cbk);
switch (req.transfer_encoding) {
.chunked => return error.ChunkedNotImplemented,
.none => return error.NotWriteable,
.content_length => |len| {
try ctx.push(cbk);
if (len < buf.len) return error.MessageTooLong;
try req.connection.?.async_writeAllDirect(buf, ctx, onWriteAll);
},
}
}
pub const FinishError = WriteError || error{MessageNotCompleted};
@@ -1757,6 +1778,7 @@ pub fn async_connectTcp(
.port = port,
.protocol = protocol,
})) |conn| {
ctx.data.conn = conn;
ctx.req.connection = conn;
return ctx.pop({});
}
@@ -1845,7 +1867,6 @@ pub fn connectTunnel(
.connection = conn,
.server_header_buffer = &buffer,
}) catch |err| {
std.log.debug("err {}", .{err});
break :tunnel err;
};
defer req.deinit();
@@ -2426,14 +2447,19 @@ pub const Ctx = struct {
pub fn pop(self: *Ctx, res: anyerror!void) !void {
if (self.stack) |stack| {
const func = stack.pop(self.alloc(), null);
const ret = @call(.auto, func, .{ self, res });
const allocator = self.alloc();
const func = stack.pop(allocator, null);
defer {
if (stack.next == null) {
allocator.destroy(stack);
self.stack = null;
self.alloc().destroy(stack);
}
return ret;
}
return @call(.auto, func, .{ self, res });
}
unreachable;
}
pub fn deinit(self: Ctx) void {
@@ -2484,62 +2510,3 @@ fn setRequestConnection(ctx: *Ctx, res: anyerror!void) anyerror!void {
ctx.req.connection = ctx.data.conn;
return ctx.pop({});
}
fn onRequestWait(ctx: *Ctx, res: anyerror!void) !void {
res catch |e| {
std.debug.print("error: {any}\n", .{e});
return e;
};
std.log.debug("REQUEST WAITED", .{});
std.log.debug("Status code: {any}", .{ctx.req.response.status});
const body = try ctx.req.reader().readAllAlloc(ctx.alloc(), 1024 * 1024);
defer ctx.alloc().free(body);
std.log.debug("Body: \n{s}", .{body});
}
fn onRequestFinish(ctx: *Ctx, res: anyerror!void) !void {
res catch |err| return err;
std.log.debug("REQUEST FINISHED", .{});
return ctx.req.async_wait(ctx, onRequestWait);
}
fn onRequestSend(ctx: *Ctx, res: anyerror!void) !void {
res catch |err| return err;
std.log.debug("REQUEST SENT", .{});
return ctx.req.async_finish(ctx, onRequestFinish);
}
pub fn onRequestConnect(ctx: *Ctx, res: anyerror!void) anyerror!void {
res catch |err| return err;
std.log.debug("REQUEST CONNECTED", .{});
return ctx.req.async_send(ctx, onRequestSend);
}
test {
const alloc = std.testing.allocator;
var loop = Loop{};
var client = Client{ .allocator = alloc };
defer client.deinit();
var req = Request{
.client = &client,
};
defer req.deinit();
var ctx = try Ctx.init(&loop, &req);
defer ctx.deinit();
var server_header_buffer: [2048]u8 = undefined;
const url = "http://www.example.com";
// const url = "http://127.0.0.1:8000/zig";
try client.async_open(
.GET,
try std.Uri.parse(url),
.{ .server_header_buffer = &server_header_buffer },
&ctx,
onRequestConnect,
);
}

4
src/run_tests.zig
View File

@@ -30,7 +30,7 @@ const xhr = @import("xhr/xhr.zig");
const storage = @import("storage/storage.zig");
const url = @import("url/url.zig");
const urlquery = @import("url/query.zig");
const Client = @import("async/Client.zig");
const Client = @import("http/async/main.zig").Client;
const documentTestExecFn = @import("dom/document.zig").testExecFn;
const HTMLDocumentTestExecFn = @import("html/document.zig").testExecFn;
@@ -86,7 +86,7 @@ fn testExecFn(
std.debug.print("documentHTMLClose error: {s}\n", .{@errorName(err)});
};
var cli = Client{ .allocator = alloc, .loop = js_env.nat_ctx.loop };
var cli = Client{ .allocator = alloc };
defer cli.deinit();
try js_env.setUserContext(.{

2
src/user_context.zig
View File

@@ -1,6 +1,6 @@
const std = @import("std");
const parser = @import("netsurf");
const Client = @import("async/Client.zig");
const Client = @import("http/async/main.zig").Client;
pub const UserContext = struct {
document: *parser.DocumentHTML,

4
src/wpt/run.zig
View File

@@ -31,7 +31,7 @@ const storage = @import("../storage/storage.zig");
const Types = @import("../main_wpt.zig").Types;
const UserContext = @import("../main_wpt.zig").UserContext;
const Client = @import("../async/Client.zig");
const Client = @import("../http/async/main.zig").Client;
// runWPT parses the given HTML file, starts a js env and run the first script
// tags containing javascript sources.
@@ -53,7 +53,7 @@ pub fn run(arena: *std.heap.ArenaAllocator, comptime dir: []const u8, f: []const
var loop = try Loop.init(alloc);
defer loop.deinit();
var cli = Client{ .allocator = alloc, .loop = &loop };
var cli = Client{ .allocator = alloc };
defer cli.deinit();
var js_env: Env = undefined;

161
src/xhr/xhr.zig
View File

@@ -32,8 +32,7 @@ const XMLHttpRequestEventTarget = @import("event_target.zig").XMLHttpRequestEven
const Mime = @import("../browser/mime.zig");
const Loop = jsruntime.Loop;
const YieldImpl = Loop.Yield(XMLHttpRequest);
const Client = @import("../async/Client.zig");
const Client = @import("../http/async/main.zig").Client;
const parser = @import("netsurf");
@@ -98,10 +97,11 @@ pub const XMLHttpRequest = struct {
proto: XMLHttpRequestEventTarget = XMLHttpRequestEventTarget{},
alloc: std.mem.Allocator,
cli: *Client,
impl: YieldImpl,
loop: Client.Loop,
priv_state: PrivState = .new,
req: ?Client.Request = null,
ctx: ?Client.Ctx = null,
method: std.http.Method,
state: u16,
@@ -135,7 +135,13 @@ pub const XMLHttpRequest = struct {
response_header_buffer: [1024 * 16]u8 = undefined,
response_status: u10 = 0,
response_override_mime_type: ?[]const u8 = null,
// TODO uncomment this field causes casting issue with
// XMLHttpRequestEventTarget. I think it's dueto an alignement issue, but
// not sure. see
// https://lightpanda.slack.com/archives/C05TRU6RBM1/p1707819010681019
// response_override_mime_type: ?[]const u8 = null,
response_mime: Mime = undefined,
response_obj: ?ResponseObj = null,
send_flag: bool = false,
@@ -288,7 +294,7 @@ pub const XMLHttpRequest = struct {
.alloc = alloc,
.headers = Headers.init(alloc),
.response_headers = Headers.init(alloc),
.impl = YieldImpl.init(loop),
.loop = Client.Loop.init(loop),
.method = undefined,
.url = null,
.uri = undefined,
@@ -320,11 +326,12 @@ pub const XMLHttpRequest = struct {
self.priv_state = .new;
if (self.req) |*r| {
r.deinit();
if (self.ctx) |*c| c.deinit();
self.ctx = null;
if (self.req) |*r| r.deinit();
self.req = null;
}
}
pub fn deinit(self: *XMLHttpRequest, alloc: std.mem.Allocator) void {
self.reset();
@@ -494,49 +501,38 @@ pub const XMLHttpRequest = struct {
log.debug("{any} {any}", .{ self.method, self.uri });
self.send_flag = true;
self.impl.yield(self);
}
// onYield is a callback called between each request's steps.
// Between each step, the code is blocking.
// Yielding allows pseudo-async and gives a chance to other async process
// to be called.
pub fn onYield(self: *XMLHttpRequest, err: ?anyerror) void {
if (err) |e| return self.onErr(e);
switch (self.priv_state) {
.new => {
self.priv_state = .open;
self.req = self.cli.open(self.method, self.uri, .{
self.req = try self.cli.create(self.method, self.uri, .{
.server_header_buffer = &self.response_header_buffer,
.extra_headers = self.headers.all(),
}) catch |e| return self.onErr(e);
},
.open => {
// prepare payload transfert.
if (self.payload) |v| self.req.?.transfer_encoding = .{ .content_length = v.len };
self.priv_state = .send;
self.req.?.send() catch |e| return self.onErr(e);
},
.send => {
if (self.payload) |payload| {
self.priv_state = .write;
self.req.?.writeAll(payload) catch |e| return self.onErr(e);
} else {
self.priv_state = .finish;
self.req.?.finish() catch |e| return self.onErr(e);
});
errdefer {
self.req.?.deinit();
self.req = null;
}
},
.write => {
self.priv_state = .finish;
self.req.?.finish() catch |e| return self.onErr(e);
},
.finish => {
self.priv_state = .wait;
self.req.?.wait() catch |e| return self.onErr(e);
},
.wait => {
self.ctx = try Client.Ctx.init(&self.loop, &self.req.?);
errdefer {
self.ctx.?.deinit();
self.ctx = null;
}
self.ctx.?.userData = self;
try self.cli.async_open(
self.method,
self.uri,
.{ .server_header_buffer = &self.response_header_buffer },
&self.ctx.?,
onRequestConnect,
);
}
fn onRequestWait(ctx: *Client.Ctx, res: anyerror!void) !void {
var self = selfCtx(ctx);
res catch |err| return self.onErr(err);
log.info("{any} {any} {d}", .{ self.method, self.uri, self.req.?.response.status });
self.priv_state = .done;
@@ -563,6 +559,7 @@ pub const XMLHttpRequest = struct {
// dispatch a progress event loadstart.
self.dispatchProgressEvent("loadstart", .{ .loaded = loaded, .total = total });
// TODO read async
const reader = self.req.?.reader();
var buffer: [1024]u8 = undefined;
var ln = buffer.len;
@@ -605,27 +602,59 @@ pub const XMLHttpRequest = struct {
self.dispatchProgressEvent("load", .{ .loaded = loaded, .total = total });
// dispatch a progress event loadend.
self.dispatchProgressEvent("loadend", .{ .loaded = loaded, .total = total });
},
.done => {
if (self.req) |*r| {
r.deinit();
if (self.ctx) |*c| c.deinit();
self.ctx = null;
if (self.req) |*r| r.deinit();
self.req = null;
}
// finalize fetch process.
return;
},
fn onRequestFinish(ctx: *Client.Ctx, res: anyerror!void) !void {
var self = selfCtx(ctx);
res catch |err| return self.onErr(err);
self.priv_state = .wait;
return ctx.req.async_wait(ctx, onRequestWait) catch |e| return self.onErr(e);
}
self.impl.yield(self);
fn onRequestSend(ctx: *Client.Ctx, res: anyerror!void) !void {
var self = selfCtx(ctx);
res catch |err| return self.onErr(err);
if (self.payload) |payload| {
self.priv_state = .write;
return ctx.req.async_writeAll(payload, ctx, onRequestWrite) catch |e| return self.onErr(e);
}
self.priv_state = .finish;
return ctx.req.async_finish(ctx, onRequestFinish) catch |e| return self.onErr(e);
}
fn onRequestWrite(ctx: *Client.Ctx, res: anyerror!void) !void {
var self = selfCtx(ctx);
res catch |err| return self.onErr(err);
self.priv_state = .finish;
return ctx.req.async_finish(ctx, onRequestFinish) catch |e| return self.onErr(e);
}
fn onRequestConnect(ctx: *Client.Ctx, res: anyerror!void) anyerror!void {
var self = selfCtx(ctx);
res catch |err| return self.onErr(err);
// prepare payload transfert.
if (self.payload) |v| self.req.?.transfer_encoding = .{ .content_length = v.len };
self.priv_state = .send;
return ctx.req.async_send(ctx, onRequestSend) catch |err| return self.onErr(err);
}
fn selfCtx(ctx: *Client.Ctx) *XMLHttpRequest {
return @ptrCast(@alignCast(ctx.userData));
}
fn onErr(self: *XMLHttpRequest, err: anyerror) void {
self.priv_state = .done;
if (self.req) |*r| {
r.deinit();
self.req = null;
}
self.err = err;
self.state = DONE;
@@ -635,6 +664,12 @@ pub const XMLHttpRequest = struct {
self.dispatchProgressEvent("loadend", .{});
log.debug("{any} {any} {any}", .{ self.method, self.uri, self.err });
if (self.ctx) |*c| c.deinit();
self.ctx = null;
if (self.req) |*r| r.deinit();
self.req = null;
}
pub fn _abort(self: *XMLHttpRequest) void {
@@ -882,7 +917,7 @@ pub fn testExecFn(
// .{ .src = "req.onload", .ex = "function cbk(event) { nb ++; evt = event; }" },
//.{ .src = "req.onload = cbk", .ex = "function cbk(event) { nb ++; evt = event; }" },
.{ .src = "req.open('GET', 'http://httpbin.io/html')", .ex = "undefined" },
.{ .src = "req.open('GET', 'https://httpbin.io/html')", .ex = "undefined" },
.{ .src = "req.setRequestHeader('User-Agent', 'lightpanda/1.0')", .ex = "undefined" },
// ensure open resets values
@@ -912,7 +947,7 @@ pub fn testExecFn(
var document = [_]Case{
.{ .src = "const req2 = new XMLHttpRequest()", .ex = "undefined" },
.{ .src = "req2.open('GET', 'http://httpbin.io/html')", .ex = "undefined" },
.{ .src = "req2.open('GET', 'https://httpbin.io/html')", .ex = "undefined" },
.{ .src = "req2.responseType = 'document'", .ex = "document" },
.{ .src = "req2.send()", .ex = "undefined" },
@@ -928,7 +963,7 @@ pub fn testExecFn(
var json = [_]Case{
.{ .src = "const req3 = new XMLHttpRequest()", .ex = "undefined" },
.{ .src = "req3.open('GET', 'http://httpbin.io/json')", .ex = "undefined" },
.{ .src = "req3.open('GET', 'https://httpbin.io/json')", .ex = "undefined" },
.{ .src = "req3.responseType = 'json'", .ex = "json" },
.{ .src = "req3.send()", .ex = "undefined" },
@@ -943,7 +978,7 @@ pub fn testExecFn(
var post = [_]Case{
.{ .src = "const req4 = new XMLHttpRequest()", .ex = "undefined" },
.{ .src = "req4.open('POST', 'http://httpbin.io/post')", .ex = "undefined" },
.{ .src = "req4.open('POST', 'https://httpbin.io/post')", .ex = "undefined" },
.{ .src = "req4.send('foo')", .ex = "undefined" },
// Each case executed waits for all loop callaback calls.
@@ -956,7 +991,7 @@ pub fn testExecFn(
var cbk = [_]Case{
.{ .src = "const req5 = new XMLHttpRequest()", .ex = "undefined" },
.{ .src = "req5.open('GET', 'http://httpbin.io/json')", .ex = "undefined" },
.{ .src = "req5.open('GET', 'https://httpbin.io/json')", .ex = "undefined" },
.{ .src = "var status = 0; req5.onload = function () { status = this.status };", .ex = "function () { status = this.status }" },
.{ .src = "req5.send()", .ex = "undefined" },