Embedded
Embedded computation-graph runtime builder (CLOACI-T-0738).
Replaces the ~60-line hand-wired main() block (four channels, an
AccumulatorContext full of Nones, a CompiledGraphFn closure, an
unused manual_rx, two tokio::spawns) that embedded CG examples used to
copy-paste. The production scheduler already does all of that wiring in
load_graph; this is the embedded-friendly face of the same machinery:
let graph = EmbeddedGraph::spawn(my_graph_declaration()).await?;
graph.push("prices", &serde_json::json!({"symbol": "X", "px": 42.0})).await?;
// ... later
graph.shutdown().await;
Manual wiring still works — this is additive.
pub
A running embedded computation graph: accumulators spawned, reactor live, events pushed via push. Dropping the value does NOT stop the graph — call shutdown.
| Name | Type | Description |
|---|---|---|
scheduler |
ComputationGraphScheduler |
|
registry |
EndpointRegistry |
|
graph_name |
String |
async
async fn spawn (decl : ComputationGraphDeclaration) -> Result < Self , String >
Wire and spawn decl (accumulators + reactor + compiled graph fn) — the whole block embedded examples used to hand-write.
Source
pub async fn spawn(decl: ComputationGraphDeclaration) -> Result<Self, String> {
let registry = EndpointRegistry::new();
let scheduler = ComputationGraphScheduler::new(registry.clone());
let graph_name = decl.name.clone();
scheduler.load_graph(decl).await?;
Ok(Self {
scheduler,
registry,
graph_name,
})
}
async
async fn push (& self , accumulator : & str , event : & impl Serialize) -> Result < () , String >
Push a JSON-serializable event into an accumulator by name (the same raw-JSON socket contract the server’s WS/REST injection uses).
Source
pub async fn push(&self, accumulator: &str, event: &impl Serialize) -> Result<(), String> {
let bytes = serde_json::to_vec(event).map_err(|e| e.to_string())?;
self.push_raw(accumulator, bytes).await
}
async
async fn push_raw (& self , accumulator : & str , bytes : Vec < u8 >) -> Result < () , String >
Push pre-encoded raw event bytes into an accumulator by name.
Source
pub async fn push_raw(&self, accumulator: &str, bytes: Vec<u8>) -> Result<(), String> {
self.registry
.send_to_accumulator(accumulator, bytes)
.await
.map(|_| ())
.map_err(|e| e.to_string())
}
fn graph_name (& self) -> & str
The graph’s name (== reactor name for self-reactor declarations).
Source
pub fn graph_name(&self) -> &str {
&self.graph_name
}
fn scheduler (& self) -> & ComputationGraphScheduler
Escape hatch: the underlying scheduler, for anything the lean surface doesn’t cover (manual force-fire, health, additional graphs).
Source
pub fn scheduler(&self) -> &ComputationGraphScheduler {
&self.scheduler
}
fn registry (& self) -> & EndpointRegistry
Escape hatch: the endpoint registry (reactor handles, health).
Source
pub fn registry(&self) -> &EndpointRegistry {
&self.registry
}
async
async fn shutdown (self)
Stop the reactor and accumulators.
Source
pub async fn shutdown(self) {
self.scheduler.shutdown_all().await;
}