Signals
Subscribe to application, router, and route events, observe listener shutdown, and use typed RPC with the in-process signal arbiter.
Signals are Tako's in-process pub/sub bus for framework-internal events and application-level RPC. The framework emits well-known signals on every request, connection, and queue job; you subscribe to the IDs that matter.
use tako::Method;
use tako::responder::Responder;
use tako::router::Router;
use tako::signals::{Signal, app_signals, ids};
use tako::types::Request;
async fn hello(_: Request) -> impl Responder { "hi" }
fn install_listeners() {
let arbiter = app_signals();
// One-shot callback when the server starts up.
arbiter.on(ids::SERVER_STARTED, |sig: Signal| async move {
println!("server.started: {:?}", sig.metadata);
});
// Long-running listener that logs every completed request.
let mut rx = arbiter.subscribe(ids::REQUEST_COMPLETED);
tokio::spawn(async move {
while let Ok(sig) = rx.recv().await {
let method = sig.metadata.get("method").cloned().unwrap_or_default();
let path = sig.metadata.get("path").cloned().unwrap_or_default();
let status = sig.metadata.get("status").cloned().unwrap_or_default();
println!("request.completed: {method} {path} -> {status}");
}
});
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
install_listeners();
let listener = tokio::net::TcpListener::bind("127.0.0.1:8080").await?;
let mut router = Router::new();
router.route(Method::GET, "/", hello);
tako::Server::builder().build().spawn_http(listener, router).result().await?;
Ok(())
}app_signals() returns the process-wide arbiter. Delivery follows the owning scope:
| Event | Recipients |
|---|---|
request.started, request.completed | Owning router and application |
route.request.started, route.request.completed | Matching route, owning router, and application |
server.started, server.stopped | Application; per-thread workers emit once per listener |
| Transport connection lifecycle events | Application |
| Custom events | The arbiter on which you explicitly emit them |
request.completed carries method, path, status, matched route, and
duration_us. Listen with router.on_signal(...) for one service or
app_signals().on(...) for all routers. server.stopped follows normal drain
completion; cancelling the entire server future skips its final event.
ROUTER_HOT_RELOAD is an application-defined convention, not an automatic event.
RPC over signals
The same arbiter doubles as a typed RPC bus:
use std::sync::Arc;
use tako::signals::SignalArbiter;
#[derive(Debug)]
struct AddRequest { a: i32, b: i32 }
#[derive(Debug, Clone)]
struct AddResponse { sum: i32 }
#[tokio::main]
async fn main() {
let arbiter = SignalArbiter::new();
arbiter.register_rpc::<AddRequest, AddResponse, _, _>(
"rpc.add",
|req: Arc<AddRequest>| async move { AddResponse { sum: req.a + req.b } },
);
let res = arbiter.call_rpc::<AddRequest, AddResponse>(
"rpc.add",
AddRequest { a: 2, b: 40 },
).await.unwrap();
assert_eq!(res.sum, 42);
}register_rpc installs a typed handler; call_rpc /
call_rpc_timeout / call_rpc_result invoke it. Errors are reported
via the RpcError enum.
Forwarding events
Register a callback or exporter on an arbiter to forward selected events.
See:
examples/signals-basicā subscribe torequest.completed,examples/signals-routeā per-router arbiter and custom IDs,examples/signals-rpcā typed RPC handlers,examples/signals-advancedandexamples/signals-complexā wildcard subscriptions and exporters.
Last updated on