agentos_v8_runtime/
stream.rs

1// Async event dispatch for child process and HTTP server streams
2
3/// Dispatch a stream event into V8 by calling the registered callback function.
4///
5/// Stream events are sent by the host when async operations (child processes,
6/// HTTP servers) produce data. The event_type determines which V8 dispatch
7/// function is called:
8/// - "child_stdout", "child_stderr", "child_exit" → _childProcessDispatch
9/// - "http_request" → _httpServerDispatch
10/// - "http2" → _http2Dispatch
11/// - "stdin", "stdin_end" → _stdinDispatch
12/// - "net_socket" → _netSocketDispatch
13/// - "signal" → __secureExecWasmSignalDispatch or _signalDispatch
14/// - "timer" → _timerDispatch
15pub fn dispatch_stream_event(scope: &mut v8::HandleScope, event_type: &str, payload: &[u8]) {
16    // Look up the dispatch function on the global object
17    let context = scope.get_current_context();
18    let global = context.global(scope);
19
20    let dispatch_names: &[&str] = match event_type {
21        "child_stdout" | "child_stderr" | "child_exit" => &["_childProcessDispatch"],
22        "http_request" => &["_httpServerDispatch"],
23        "http2" => &["_http2Dispatch"],
24        "stdin" | "stdin_end" => &["_stdinDispatch"],
25        "net_socket" => &["_netSocketDispatch"],
26        "signal" => &["__secureExecWasmSignalDispatch", "_signalDispatch"],
27        "timer" => &["_timerDispatch"],
28        _ => return, // Unknown event type — ignore
29    };
30
31    for dispatch_name in dispatch_names {
32        let key = v8::String::new(scope, dispatch_name).unwrap();
33        let maybe_fn = global.get(scope, key.into());
34
35        if let Some(func_val) = maybe_fn {
36            if func_val.is_function() {
37                let func = v8::Local::<v8::Function>::try_from(func_val).unwrap();
38
39                // Pass event_type and payload as arguments.
40                let event_str = v8::String::new(scope, event_type).unwrap();
41                let payload_val = if !payload.is_empty() {
42                    let maybe_v8_payload = {
43                        let tc = &mut v8::TryCatch::new(scope);
44                        crate::bridge::deserialize_v8_value(tc, payload).ok()
45                    };
46                    match maybe_v8_payload {
47                        Some(v) => v,
48                        None => match std::str::from_utf8(payload) {
49                            Ok(text) => match v8::String::new(scope, text) {
50                                Some(json_text) => v8::json::parse(scope, json_text)
51                                    .unwrap_or_else(|| json_text.into()),
52                                None => v8::null(scope).into(),
53                            },
54                            Err(_) => v8::null(scope).into(),
55                        },
56                    }
57                } else {
58                    v8::null(scope).into()
59                };
60
61                let undefined = v8::undefined(scope);
62                let args: &[v8::Local<v8::Value>] = &[event_str.into(), payload_val];
63                func.call(scope, undefined.into(), args);
64                return;
65            }
66        }
67    }
68}
agentos_v8_runtime/stream.rs

agentos_v8_runtime/
stream.rs
