broker: tap/untap = receive a copy of every routed message
ci / check (push) Successful in 12m13s

Tapped copies {type:tapped, message:{id,from,to,reply_to,body}, ts} are
delivered directly (never tapped themselves), dropped when the tapper's
mailbox is full, and the tap is saved in actor.json like subscriptions.
This commit is contained in:
Sausage Agent
2026-10-16 11:14:09 +00:00
parent 2ff30ecf70
commit 49453fa9af
4 changed files with 182 additions and 17 deletions
+46 -14
View File
@@ -1,6 +1,7 @@
//! The broker: message routing, mailboxes, subscriptions, the version store,
//! and the requests to `broker` (spawn, upgrade, rollback, stop, start,
//! subscribe, unsubscribe, scan, exit). `supervisor.rs` runs each WASM actor.
//! The broker: message routing, mailboxes, subscriptions, taps, the version
//! store, and the requests to `broker` (spawn, upgrade, rollback, stop, start,
//! subscribe, unsubscribe, tap, untap, scan, exit). `supervisor.rs` runs each
//! WASM actor.
//!
//! The broker accepts only `.wasm` files. It does not know any source language.
//! The world folder is the source of truth: `scan` boots every
@@ -133,17 +134,23 @@ impl Broker {
Some(own) => format!("{from}/{own}"),
None => format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)),
};
let recipients: Vec<String> = match to {
Some(to) => vec![to.to_string()],
None => self
.metas
.lock()
.unwrap()
.iter()
.filter(|(_, m)| m.subscriptions.contains(from))
.map(|(id, _)| id.clone())
.collect(),
let (recipients, tappers) = {
let metas = self.metas.lock().unwrap();
let recipients: Vec<String> = match to {
Some(to) => vec![to.to_string()],
None => metas
.iter()
.filter(|(_, m)| m.subscriptions.contains(from))
.map(|(id, _)| id.clone())
.collect(),
};
let tappers: Vec<String> = metas.iter().filter(|(_, m)| m.tapped).map(|(id, _)| id.clone()).collect();
(recipients, tappers)
};
if !tappers.is_empty() {
let message = json!({"id": id, "from": from, "to": to, "reply_to": reply_to, "body": body});
self.tap(&tappers, message);
}
for to in recipients {
self.deliver(Envelope {
id: id.clone(),
@@ -156,6 +163,25 @@ impl Broker {
id
}
/// Give every tapper a copy of one routed message. Copies skip `send`, so
/// they are never tapped themselves; a full mailbox drops the copy.
fn tap(&self, tappers: &[String], message: Value) {
let ts = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_millis() as u64);
let body = json!({"type": "tapped", "message": message, "ts": ts});
let mailboxes = self.mailboxes.lock().unwrap();
for tapper in tappers {
if let Some(tx) = mailboxes.get(tapper) {
let _ = tx.try_send(Envelope {
id: format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)),
from: "broker".into(),
to: tapper.clone(),
reply_to: None,
body: body.clone(),
});
}
}
}
/// Route one stdout line of a WASM actor.
pub fn route_line(&self, from: &str, line: &str) {
let line = line.trim();
@@ -467,6 +493,12 @@ impl Broker {
})?;
Ok(json!({"type": format!("{ty}d"), "actor": publisher}))
}
"tap" | "untap" => {
let on = ty == "tap";
self.update_meta(&env.from, |m| m.tapped = on)?;
let answer = if on { "tapping" } else { "untapped" };
Ok(json!({"type": answer}))
}
"scan" => Ok(json!({"type": "scanned", "changes": self.scan().await})),
"exit" => {
self.log("broker", &format!("exit requested by {}", env.from));
@@ -474,7 +506,7 @@ impl Broker {
self.exit.notify_one();
Ok(json!({"type": "exiting"}))
}
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, subscribe, unsubscribe, scan, exit"),
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, subscribe, unsubscribe, tap, untap, scan, exit"),
}
}