104 lines
3.8 KiB
Zig
104 lines
3.8 KiB
Zig
/// Parent-side endpoint actor.
|
|
///
|
|
/// Wraps a subprocess, accumulates framed CBOR from its stdout, and dispatches
|
|
/// decoded send_named messages to named local actors via the env proc table.
|
|
///
|
|
/// Local actors send outbound messages to the endpoint with:
|
|
/// {"send", from_id: u64, to_name: text, payload: raw_cbor}
|
|
const std = @import("std");
|
|
const tp = @import("thespian");
|
|
const cbor = @import("cbor");
|
|
const framing = @import("framing");
|
|
const protocol = @import("protocol");
|
|
|
|
const subprocess = tp.subprocess;
|
|
|
|
const proc_tag = "endpoint_proc";
|
|
|
|
pub const Args = struct {
|
|
allocator: std.mem.Allocator,
|
|
/// CBOR-encoded argv array for the child binary.
|
|
/// Must be heap-allocated; endpoint.init will free it.
|
|
argv: []const u8,
|
|
};
|
|
|
|
const Endpoint = struct {
|
|
allocator: std.mem.Allocator,
|
|
proc: subprocess,
|
|
accumulator: framing.Accumulator,
|
|
receiver: tp.Receiver(*@This()),
|
|
|
|
fn start(args: Args) tp.result {
|
|
return init(args) catch |e| return tp.exit_error(e, @errorReturnTrace());
|
|
}
|
|
|
|
fn init(args: Args) !void {
|
|
defer args.allocator.free(args.argv);
|
|
const proc = try subprocess.init(args.allocator, tp.message{ .buf = args.argv }, proc_tag, .Pipe);
|
|
const self = try args.allocator.create(@This());
|
|
self.* = .{
|
|
.allocator = args.allocator,
|
|
.proc = proc,
|
|
.accumulator = .{},
|
|
.receiver = .init(receive_fn, deinit, self),
|
|
};
|
|
errdefer self.deinit();
|
|
tp.receive(&self.receiver);
|
|
}
|
|
|
|
fn deinit(self: *@This()) void {
|
|
self.proc.deinit();
|
|
self.allocator.destroy(self);
|
|
}
|
|
|
|
fn receive_fn(self: *@This(), from: tp.pid_ref, m: tp.message) tp.result {
|
|
return self.receive(from, m) catch |e| return tp.exit_error(e, @errorReturnTrace());
|
|
}
|
|
|
|
fn receive(self: *@This(), _: tp.pid_ref, m: tp.message) !void {
|
|
var bytes: []const u8 = "";
|
|
var from_id: u64 = 0;
|
|
var to_name: []const u8 = "";
|
|
var payload: []const u8 = "";
|
|
if (try m.match(.{ proc_tag, "stdout", tp.extract(&bytes) })) {
|
|
if (self.accumulator.feed(bytes)) |frame| try self.dispatch_inbound(frame);
|
|
} else if (try m.match(.{ proc_tag, "stderr", tp.any })) {
|
|
} else if (try m.match(.{ "send", tp.extract(&from_id), tp.extract(&to_name), cbor.extract_cbor(&payload) })) {
|
|
try self.send_wire(from_id, to_name, payload);
|
|
} else if (try m.match(.{ proc_tag, "term", "exited", tp.any })) {
|
|
return tp.exit("transport_closed");
|
|
} else if (try m.match(.{ proc_tag, "term", tp.any, tp.any })) {
|
|
return tp.exit("transport_closed");
|
|
} else {
|
|
return tp.unexpected(m);
|
|
}
|
|
}
|
|
|
|
fn dispatch_inbound(_: *@This(), frame: []const u8) !void {
|
|
const msg = try protocol.decode(frame);
|
|
switch (msg) {
|
|
.send => |s| {
|
|
_ = s;
|
|
return tp.exit_error(error.UnexpectedSend, null);
|
|
},
|
|
.send_named => |s| {
|
|
const actor = tp.env.get().proc(s.to_name);
|
|
try actor.send_raw(tp.message{ .buf = s.payload });
|
|
},
|
|
.link, .exit, .proxy_id, .transport_error => return tp.exit_error(error.UnexpectedMessage, null),
|
|
}
|
|
}
|
|
|
|
fn send_wire(self: *@This(), from_id: u64, to_name: []const u8, payload: []const u8) !void {
|
|
var msg_buf: [framing.max_frame_size]u8 = undefined;
|
|
var msg_stream: std.Io.Writer = .fixed(&msg_buf);
|
|
try protocol.encode_send_named(&msg_stream, from_id, to_name, payload);
|
|
|
|
var frame_buf: [framing.max_frame_size + 4]u8 = undefined;
|
|
var frame_stream: std.Io.Writer = .fixed(&frame_buf);
|
|
try framing.write_frame(&frame_stream, msg_stream.buffered());
|
|
try self.proc.send(frame_stream.buffered());
|
|
}
|
|
};
|
|
|
|
pub const start = Endpoint.start;
|