neurocyte/thespian
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

WIP: implement proxy

Browse source
This commit is contained in:
CJ van den Berg 2026-03-07 15:08:52 +01:00
parent ae619675de
commit d40db53545
Signed by: neurocyte
GPG key ID: 8EB1E1BB660E3FB9
3 changed files with 104 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

23
src/remote/endpoint.zig
View file

@ -10,6 +10,7 @@ const tp = @import("thespian");
const cbor = @import("cbor");
const framing = @import("framing");
const protocol = @import("protocol");
const proxy = @import("proxy");
const subprocess = tp.subprocess;
@ -58,13 +59,20 @@ const Endpoint = struct {
fn receive(self: *@This(), _: tp.pid_ref, m: tp.message) !void {
var bytes: []const u8 = "";
var from_id: u64 = 0;
var to_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_id), cbor.extract_cbor(&payload) })) {
// Outbound send by remote ID — from a local proxy.
try self.send_wire_by_id(from_id, to_id, payload);
} 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);
// Outbound send_named — from a local actor addressing a well-known remote actor by name.
try self.send_wire_named(from_id, to_name, payload);
} else if (try m.match(.{ "proxy_exit", tp.any, tp.any })) {
// A local proxy has exited; remove it from the proxy table (stub until table is implemented).
} 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 })) {
@ -89,7 +97,18 @@ const Endpoint = struct {
}
}
fn send_wire(self: *@This(), from_id: u64, to_name: []const u8, payload: []const u8) !void {
fn send_wire_by_id(self: *@This(), from_id: u64, to_id: u64, 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(&msg_stream, from_id, to_id, 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());
}
fn send_wire_named(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);