From c0b003a5b19fac78d88d9e44936a305a722180c2 Mon Sep 17 00:00:00 2001 From: Liam Carvajal Date: Fri, 31 Jul 2026 22:47:03 +0200 Subject: [PATCH] refactor(core): move heartbeat/liveness logic into core behind remote-adapter feature Moves the duplicated heartbeat/liveness logic from the postgres and mongodb adapters into socketioxide-core behind the existing remote-adapter feature flag. - HeartbeatTracker: wraps nodes_liveness + hb_timeout; on_heartbeat (update-or-push, returns whether an InitHeartbeat reply is due), server_count (prune dead nodes + self), is_alive - HeartbeatSender trait: uid, send_req, with default emit_heartbeat, emit_init_heartbeat and recv_heartbeat implementations - heartbeat_loop helper Postgres and mongodb now keep only a thin HeartbeatSender impl. No behavior or wire-format change. Closes #767 --- .../src/adapter/heartbeat.rs | 228 ++++++++++++++++++ crates/socketioxide-core/src/adapter/mod.rs | 2 + crates/socketioxide-mongodb/src/lib.rs | 106 ++------ crates/socketioxide-postgres/src/lib.rs | 195 ++++++--------- 4 files changed, 325 insertions(+), 206 deletions(-) create mode 100644 crates/socketioxide-core/src/adapter/heartbeat.rs diff --git a/crates/socketioxide-core/src/adapter/heartbeat.rs b/crates/socketioxide-core/src/adapter/heartbeat.rs new file mode 100644 index 00000000..eecaef5f --- /dev/null +++ b/crates/socketioxide-core/src/adapter/heartbeat.rs @@ -0,0 +1,228 @@ +//! Shared heartbeat/liveness logic for remote adapters. + +use std::{ + future::Future, + sync::{Arc, Mutex}, + time::{Duration, Instant}, +}; + +use crate::{ + Uid, + adapter::remote_packet::{RequestOut, RequestTypeIn, RequestTypeOut}, +}; + +/// Tracks the liveness of remote nodes based on the heartbeats they send. +pub struct HeartbeatTracker { + /// The last time each remote node was seen alive. + nodes_liveness: Mutex>, + /// The duration after which a node is considered dead. + hb_timeout: Duration, +} + +impl HeartbeatTracker { + /// Create a new `HeartbeatTracker` with the given heartbeat timeout. + pub fn new(hb_timeout: Duration) -> Self { + Self { + nodes_liveness: Mutex::new(Vec::new()), + hb_timeout, + } + } + + /// Record a heartbeat received from `origin`. + /// + /// Returns `true` if the node is new and an [`RequestTypeIn::InitHeartbeat`] + /// reply is due, `false` if the node was already known. + pub fn on_heartbeat(&self, origin: Uid) -> bool { + let mut nodes_liveness = self.nodes_liveness.lock().unwrap(); + for (id, liveness) in nodes_liveness.iter_mut() { + if *id == origin { + *liveness = Instant::now(); + return false; + } + } + nodes_liveness.push((origin, Instant::now())); + true + } + + /// The number of live nodes, including the current one. + /// + /// Dead nodes are pruned from the tracker. + pub fn server_count(&self) -> u16 { + let threshold = Instant::now() - self.hb_timeout; + let mut nodes_liveness = self.nodes_liveness.lock().unwrap(); + nodes_liveness.retain(|(_, v)| v > &threshold); + (nodes_liveness.len() + 1) as u16 + } + + /// Whether the given remote node is currently considered alive. + pub fn is_alive(&self, uid: Uid) -> bool { + let threshold = Instant::now() - self.hb_timeout; + self.nodes_liveness + .lock() + .unwrap() + .iter() + .any(|(id, liveness)| *id == uid && *liveness > threshold) + } +} + +/// The sending side of the heartbeat protocol, implemented by the remote adapters. +pub trait HeartbeatSender: Send + Sync + 'static { + /// The error type returned by the adapter. + type Error: std::error::Error + Send + 'static; + + /// The unique identifier of the current node. + fn uid(&self) -> Uid; + + /// Send a request to a specific target node or broadcast it to all nodes + /// if no target is specified. + fn send_req( + &self, + req: RequestOut<'_>, + target: Option, + ) -> impl Future> + Send; + + /// Emit a heartbeat to the specified target node or broadcast it to all nodes. + fn emit_heartbeat( + &self, + target: Option, + ) -> impl Future> + Send { + async move { + self.send_req( + RequestOut::new_empty(self.uid(), RequestTypeOut::Heartbeat), + target, + ) + .await + } + } + + /// Emit an initial heartbeat to all nodes. + fn emit_init_heartbeat(&self) -> impl Future> + Send { + async move { + self.send_req( + RequestOut::new_empty(self.uid(), RequestTypeOut::InitHeartbeat), + None, + ) + .await + } + } + + /// Receive a heartbeat from a remote node. + /// + /// It might be an [`RequestTypeIn::InitHeartbeat`] packet, in which case a + /// heartbeat is re-emitted to the new node so it discovers the current one. + fn recv_heartbeat( + self: &Arc, + tracker: &HeartbeatTracker, + req_type: RequestTypeIn, + origin: Uid, + ) { + tracing::debug!(?req_type, "{:?} received", req_type); + if tracker.on_heartbeat(origin) && matches!(req_type, RequestTypeIn::InitHeartbeat) { + tracing::debug!( + ?origin, + "initial heartbeat detected, saying hello to the new node" + ); + + let this = self.clone(); + tokio::spawn(async move { + if let Err(err) = this.emit_heartbeat(Some(origin)).await { + tracing::warn!( + "could not re-emit heartbeat after new node detection: {:?}", + err + ); + } + }); + } + } +} + +/// Run the heartbeat loop, emitting a heartbeat to all nodes every `hb_interval`. +/// +/// The first tick is skipped as it yields immediately. +pub async fn heartbeat_loop( + adapter: Arc, + hb_interval: Duration, +) -> Result<(), S::Error> { + let mut interval = tokio::time::interval(hb_interval); + interval.tick().await; // first tick yields immediately + loop { + interval.tick().await; + adapter.emit_heartbeat(None).await?; + } +} + +#[cfg(test)] +mod tests { + use std::{sync::Arc, time::Duration}; + + use super::*; + + struct TestSender { + uid: Uid, + } + + impl HeartbeatSender for TestSender { + type Error = std::io::Error; + + fn uid(&self) -> Uid { + self.uid + } + + async fn send_req( + &self, + _req: RequestOut<'_>, + _target: Option, + ) -> Result<(), Self::Error> { + Ok(()) + } + } + + #[test] + fn on_heartbeat_new_node() { + let tracker = HeartbeatTracker::new(Duration::from_secs(60)); + assert!(tracker.on_heartbeat(Uid::new())); + } + + #[test] + fn on_heartbeat_known_node() { + let tracker = HeartbeatTracker::new(Duration::from_secs(60)); + let uid = Uid::new(); + assert!(tracker.on_heartbeat(uid)); + assert!(!tracker.on_heartbeat(uid)); + } + + #[test] + fn server_count_includes_self() { + let tracker = HeartbeatTracker::new(Duration::from_secs(60)); + assert_eq!(tracker.server_count(), 1); + tracker.on_heartbeat(Uid::new()); + tracker.on_heartbeat(Uid::new()); + assert_eq!(tracker.server_count(), 3); + } + + #[test] + fn dead_nodes_are_pruned() { + let tracker = HeartbeatTracker::new(Duration::from_millis(50)); + tracker.on_heartbeat(Uid::new()); + std::thread::sleep(Duration::from_millis(60)); + assert_eq!(tracker.server_count(), 1); + } + + #[test] + fn is_alive_reflects_timeout() { + let tracker = HeartbeatTracker::new(Duration::from_millis(50)); + let uid = Uid::new(); + assert!(!tracker.is_alive(uid)); + tracker.on_heartbeat(uid); + assert!(tracker.is_alive(uid)); + std::thread::sleep(Duration::from_millis(60)); + assert!(!tracker.is_alive(uid)); + } + + #[tokio::test] + async fn recv_heartbeat_emits_reply_for_new_node() { + let adapter = Arc::new(TestSender { uid: Uid::new() }); + let tracker = HeartbeatTracker::new(Duration::from_secs(60)); + adapter.recv_heartbeat(&tracker, RequestTypeIn::InitHeartbeat, Uid::new()); + } +} diff --git a/crates/socketioxide-core/src/adapter/mod.rs b/crates/socketioxide-core/src/adapter/mod.rs index 55b3014e..8e3f6089 100644 --- a/crates/socketioxide-core/src/adapter/mod.rs +++ b/crates/socketioxide-core/src/adapter/mod.rs @@ -26,6 +26,8 @@ use errors::{AdapterError, BroadcastError, SocketError}; pub mod errors; #[cfg(feature = "remote-adapter")] +pub mod heartbeat; +#[cfg(feature = "remote-adapter")] pub mod remote_packet; #[cfg(feature = "remote-adapter")] pub mod stream; diff --git a/crates/socketioxide-mongodb/src/lib.rs b/crates/socketioxide-mongodb/src/lib.rs index 6a6a8c6c..9d253ee2 100644 --- a/crates/socketioxide-mongodb/src/lib.rs +++ b/crates/socketioxide-mongodb/src/lib.rs @@ -95,7 +95,7 @@ use std::{ pin::Pin, sync::{Arc, Mutex}, task::{Context, Poll}, - time::{Duration, Instant}, + time::Duration, }; use futures_core::{Stream, future::Future}; @@ -110,6 +110,7 @@ use socketioxide_core::{ adapter::{ BroadcastOptions, CoreAdapter, CoreLocalAdapter, DefinedAdapter, RemoteSocketData, Room, RoomParam, SocketEmitter, Spawnable, + heartbeat::{HeartbeatSender, HeartbeatTracker, heartbeat_loop}, stream::{AckStream, ChanStream, DropStream, ResponseHandlers}, }, packet::Packet, @@ -319,8 +320,8 @@ pub struct CustomMongoDbAdapter { uid: Uid, /// The local adapter, used to manage local rooms and socket stores. local: CoreLocalAdapter, - /// A map of nodes liveness, with the last time remote nodes were seen alive. - nodes_liveness: Mutex>, + /// Tracks the liveness of remote nodes based on the heartbeats they send. + heartbeat: HeartbeatTracker, /// A map of response handlers used to await for responses from the remote servers. responses: Arc>>, } @@ -339,7 +340,7 @@ impl CoreAdapter for CustomMongoDbAdapter uid, driver: state.driver.clone(), config: state.config.clone(), - nodes_liveness: Mutex::new(Vec::new()), + heartbeat: HeartbeatTracker::new(state.config.hb_timeout), responses: Arc::new(Mutex::new(HashMap::new())), } } @@ -348,7 +349,7 @@ impl CoreAdapter for CustomMongoDbAdapter let fut = async move { let stream = self.driver.watch(self.uid, self.local.path()).await?; tokio::spawn(self.clone().handle_ev_stream(stream)); - tokio::spawn(self.clone().heartbeat_job()); + tokio::spawn(heartbeat_loop(self.clone(), self.config.hb_interval)); // Send initial heartbeat when starting. self.emit_init_heartbeat().await.map_err(|e| match e { @@ -368,10 +369,7 @@ impl CoreAdapter for CustomMongoDbAdapter /// Get the number of servers by iterating over the node liveness heartbeats. async fn server_count(&self) -> Result { - let treshold = std::time::Instant::now() - self.config.hb_timeout; - let mut nodes_liveness = self.nodes_liveness.lock().unwrap(); - nodes_liveness.retain(|(_, v)| v > &treshold); - Ok((nodes_liveness.len() + 1) as u16) + Ok(self.heartbeat.server_count()) } /// Broadcast a packet to all the servers to send them through their sockets. @@ -551,15 +549,6 @@ impl CoreAdapter for CustomMongoDbAdapter } impl CustomMongoDbAdapter { - async fn heartbeat_job(self: Arc) -> Result<(), Error> { - let mut interval = tokio::time::interval(self.config.hb_interval); - interval.tick().await; // first tick yields immediately - loop { - interval.tick().await; - self.emit_heartbeat(None).await?; - } - } - async fn handle_ev_stream( self: Arc, mut stream: impl Stream> + Unpin, @@ -624,7 +613,7 @@ impl CustomMongoDbAdapter { self.recv_fetch_sockets(req.node_id, req.id, opts) } req_type @ (RequestTypeIn::Heartbeat | RequestTypeIn::InitHeartbeat, _) => { - self.recv_heartbeat(req_type.0, req.node_id) + self.recv_heartbeat(&self.heartbeat, req_type.0, req.node_id) } _ => (), } @@ -734,49 +723,6 @@ impl CustomMongoDbAdapter { }); } - /// Receive a heartbeat from a remote node. - /// It might be a FirstHeartbeat packet, in which case we are re-emitting a heartbeat to the remote node. - fn recv_heartbeat(self: &Arc, req_type: RequestTypeIn, origin: Uid) { - tracing::debug!(?req_type, "{:?} received", req_type); - let mut node_liveness = self.nodes_liveness.lock().unwrap(); - // Even with a FirstHeartbeat packet we first consume the node liveness to - // ensure that the node is not already in the list. - for (id, liveness) in node_liveness.iter_mut() { - if *id == origin { - *liveness = Instant::now(); - return; - } - } - - node_liveness.push((origin, Instant::now())); - - if matches!(req_type, RequestTypeIn::InitHeartbeat) { - tracing::debug!( - ?origin, - "initial heartbeat detected, saying hello to the new node" - ); - - let this = self.clone(); - tokio::spawn(async move { - if let Err(err) = this.emit_heartbeat(Some(origin)).await { - tracing::warn!( - "could not re-emit heartbeat after new node detection: {:?}", - err - ); - } - }); - } - } - - /// Send a request to a specific target node or broadcast it to all nodes if no target is specified. - async fn send_req(&self, req: RequestOut<'_>, target: Option) -> Result<(), Error> { - tracing::trace!(?req, "sending request"); - let head = ItemHeader::Req { target }; - let req = self.new_packet(head, &req)?; - self.driver.emit(&req).await.map_err(Error::from_driver)?; - Ok(()) - } - /// Send a response to the node that sent the request. fn send_res( &self, @@ -832,25 +778,6 @@ impl CustomMongoDbAdapter { Ok(stream) } - /// Emit a heartbeat to the specified target node or broadcast to all nodes. - async fn emit_heartbeat(&self, target: Option) -> Result<(), Error> { - // Send heartbeat when starting. - self.send_req( - RequestOut::new_empty(self.uid, RequestTypeOut::Heartbeat), - target, - ) - .await - } - - /// Emit an initial heartbeat to all nodes. - async fn emit_init_heartbeat(&self) -> Result<(), Error> { - // Send initial heartbeat when starting. - self.send_req( - RequestOut::new_empty(self.uid, RequestTypeOut::InitHeartbeat), - None, - ) - .await - } fn new_packet(&self, head: ItemHeader, data: &impl Serialize) -> Result> { let ns = &self.local.path(); let uid = self.uid; @@ -862,6 +789,23 @@ impl CustomMongoDbAdapter { } } +impl HeartbeatSender for CustomMongoDbAdapter { + type Error = Error; + + fn uid(&self) -> Uid { + self.uid + } + + /// Send a request to a specific target node or broadcast it to all nodes if no target is specified. + async fn send_req(&self, req: RequestOut<'_>, target: Option) -> Result<(), Error> { + tracing::trace!(?req, "sending request"); + let head = ItemHeader::Req { target }; + let req = self.new_packet(head, &req)?; + self.driver.emit(&req).await.map_err(Error::from_driver)?; + Ok(()) + } +} + /// The result of the init future. #[must_use = "futures do nothing unless you `.await` or poll them"] pub struct InitRes(futures_core::future::BoxFuture<'static, Result<(), D::Error>>); diff --git a/crates/socketioxide-postgres/src/lib.rs b/crates/socketioxide-postgres/src/lib.rs index 76e7393c..f956e421 100644 --- a/crates/socketioxide-postgres/src/lib.rs +++ b/crates/socketioxide-postgres/src/lib.rs @@ -63,6 +63,7 @@ use socketioxide_core::{ BroadcastOptions, CoreAdapter, CoreLocalAdapter, DefinedAdapter, RemoteSocketData, Room, RoomParam, SocketEmitter, Spawnable, errors::{AdapterError, BroadcastError}, + heartbeat::{HeartbeatSender, HeartbeatTracker, heartbeat_loop}, remote_packet::{ RequestIn, RequestOut, RequestTypeIn, RequestTypeOut, Response, ResponseType, ResponseTypeId, @@ -78,7 +79,7 @@ use std::{ pin::Pin, sync::{Arc, Mutex, OnceLock}, task::{Context, Poll}, - time::{Duration, Instant}, + time::Duration, }; use tokio::{sync::mpsc, task::AbortHandle}; @@ -343,8 +344,8 @@ pub struct CustomPostgresAdapter { config: PostgresAdapterConfig, /// The local adapter, used to manage local rooms and socket stores. local: CoreLocalAdapter, - /// A map of nodes liveness, with the last time remote nodes were seen alive. - nodes_liveness: Mutex>, + /// Tracks the liveness of remote nodes based on the heartbeats they send. + heartbeat: HeartbeatTracker, /// A map of response handlers used to await for responses from the remote servers. responses: Arc>>, @@ -366,7 +367,7 @@ impl CoreAdapter for CustomPostgresAdapter local, driver: state.driver.clone(), config: state.config.clone(), - nodes_liveness: Mutex::new(Vec::new()), + heartbeat: HeartbeatTracker::new(state.config.hb_timeout), responses: Arc::new(Mutex::new(HashMap::new())), ev_stream_task: OnceLock::new(), hb_task: OnceLock::new(), @@ -394,7 +395,8 @@ impl CoreAdapter for CustomPostgresAdapter self.ev_stream_task.set(ev_stream_task).is_ok(), "Adapter::init should be called only once" ); - let hb_task = tokio::spawn(self.clone().heartbeat_job()).abort_handle(); + let hb_task = + tokio::spawn(heartbeat_loop(self.clone(), self.config.hb_interval)).abort_handle(); assert!( self.hb_task.set(hb_task).is_ok(), "Adapter::init should be called only once" @@ -433,10 +435,7 @@ impl CoreAdapter for CustomPostgresAdapter /// Get the number of servers by iterating over the node liveness heartbeats. async fn server_count(&self) -> Result { - let treshold = std::time::Instant::now() - self.config.hb_timeout; - let mut nodes_liveness = self.nodes_liveness.lock().unwrap(); - nodes_liveness.retain(|(_, v)| v > &treshold); - Ok((nodes_liveness.len() + 1) as u16) + Ok(self.heartbeat.server_count()) } /// Broadcast a packet to all the servers to send them through their sockets. @@ -647,15 +646,6 @@ impl CoreAdapter for CustomPostgresAdapter } impl CustomPostgresAdapter { - async fn heartbeat_job(self: Arc) -> Result<(), Error> { - let mut interval = tokio::time::interval(self.config.hb_interval); - interval.tick().await; // first tick yields immediately - loop { - interval.tick().await; - self.emit_heartbeat(None).await?; - } - } - async fn cleanup_job(self: Arc) { let mut interval = tokio::time::interval(self.config.cleanup_interval); interval.tick().await; // first tick yields immediately @@ -780,7 +770,7 @@ impl CustomPostgresAdapter { self.recv_fetch_sockets(req.node_id, req.id, opts) } req_type @ (RequestTypeIn::Heartbeat | RequestTypeIn::InitHeartbeat, _) => { - self.recv_heartbeat(req_type.0, req.node_id) + self.recv_heartbeat(&self.heartbeat, req_type.0, req.node_id) } _ => (), } @@ -889,93 +879,6 @@ impl CustomPostgresAdapter { }); } - /// Receive a heartbeat from a remote node. - /// It might be a FirstHeartbeat packet, in which case we are re-emitting a heartbeat to the remote node. - fn recv_heartbeat(self: &Arc, req_type: RequestTypeIn, origin: Uid) { - tracing::debug!(?req_type, "{:?} received", req_type); - let mut node_liveness = self.nodes_liveness.lock().unwrap(); - // Even with a FirstHeartbeat packet we first consume the node liveness to - // ensure that the node is not already in the list. - for (id, liveness) in node_liveness.iter_mut() { - if *id == origin { - *liveness = Instant::now(); - return; - } - } - - node_liveness.push((origin, Instant::now())); - - if matches!(req_type, RequestTypeIn::InitHeartbeat) { - tracing::debug!( - ?origin, - "initial heartbeat detected, saying hello to the new node" - ); - - let this = self.clone(); - tokio::spawn(async move { - if let Err(err) = this.emit_heartbeat(Some(origin)).await { - tracing::warn!( - "could not re-emit heartbeat after new node detection: {:?}", - err - ); - } - }); - } - } - - /// Send a request to a specific target node or broadcast it to all nodes if no target is specified. - /// - /// The request body is serialized as JSON and wrapped in a [`RequestPacket`]. When the - /// serialized body exceeds [`PostgresAdapterConfig::payload_threshold`], it is stored in the - /// attachment table and only the row id travels over NOTIFY. - async fn send_req(&self, req: RequestOut<'_>, target: Option) -> Result<(), Error> { - tracing::trace!(?req, "sending request"); - let chan = match target { - Some(target) => self.get_node_chan(target), - None => self.get_global_chan(), - }; - - let node_id = self.local.server_id(); - let is_binary = req.is_binary(); - let body = if is_binary { - rmp_serde::to_vec(&req)? - } else { - serde_json::to_vec(&req)? - }; - - let payload = if body.len() >= self.config.payload_threshold || is_binary { - let id = self - .driver - .push_attachment(&self.config.table_name, &body) - .await - .map_err(Error::Driver)?; - - tracing::debug!("pushed attachment {id} for req {}", req.id); - - serde_json::to_string(&RequestPacket::<()>::RequestWithAttachment { - node_id, - is_binary, - id, - })? - } else { - assert!( - !is_binary, - "binary packets should be stored in attachment table and serialized in msgpack" - ); - - let body = - String::from_utf8(body).expect("this should be serde_json output and valid UTF8"); - let payload = RawValue::from_string(body)?; - serde_json::to_string(&RequestPacket::Request { node_id, payload })? - }; - - self.driver - .notify(&chan, &payload) - .await - .map_err(Error::Driver)?; - Ok(()) - } - /// Send a response to the node that sent the request. /// /// When the serialized response exceeds [`PostgresAdapterConfig::payload_threshold`], it is @@ -1073,25 +976,6 @@ impl CustomPostgresAdapter { Ok(stream) } - /// Emit a heartbeat to the specified target node or broadcast to all nodes. - async fn emit_heartbeat(&self, target: Option) -> Result<(), Error> { - self.send_req( - RequestOut::new_empty(self.local.server_id(), RequestTypeOut::Heartbeat), - target, - ) - .await - } - - /// Emit an initial heartbeat to all nodes. - async fn emit_init_heartbeat(&self) -> Result<(), Error> { - // Send initial heartbeat when starting. - self.send_req( - RequestOut::new_empty(self.local.server_id(), RequestTypeOut::InitHeartbeat), - None, - ) - .await - } - // == All channels are hashed to avoid thresspassing the 63 bytes limit on postgres channel == // We cannot constraint the length of the channel name because it is generated dynamically. @@ -1114,6 +998,67 @@ impl CustomPostgresAdapter { } } +impl HeartbeatSender for CustomPostgresAdapter { + type Error = Error; + + fn uid(&self) -> Uid { + self.local.server_id() + } + + /// Send a request to a specific target node or broadcast it to all nodes if no target is specified. + /// + /// The request body is serialized as JSON and wrapped in a [`RequestPacket`]. When the + /// serialized body exceeds [`PostgresAdapterConfig::payload_threshold`], it is stored in the + /// attachment table and only the row id travels over NOTIFY. + async fn send_req(&self, req: RequestOut<'_>, target: Option) -> Result<(), Error> { + tracing::trace!(?req, "sending request"); + let chan = match target { + Some(target) => self.get_node_chan(target), + None => self.get_global_chan(), + }; + + let node_id = self.local.server_id(); + let is_binary = req.is_binary(); + let body = if is_binary { + rmp_serde::to_vec(&req)? + } else { + serde_json::to_vec(&req)? + }; + + let payload = if body.len() >= self.config.payload_threshold || is_binary { + let id = self + .driver + .push_attachment(&self.config.table_name, &body) + .await + .map_err(Error::Driver)?; + + tracing::debug!("pushed attachment {id} for req {}", req.id); + + serde_json::to_string(&RequestPacket::<()>::RequestWithAttachment { + node_id, + is_binary, + id, + })? + } else { + assert!( + !is_binary, + "binary packets should be stored in attachment table and serialized in msgpack" + ); + + let body = + String::from_utf8(body).expect("this should be serde_json output and valid UTF8"); + let payload = RawValue::from_string(body)?; + serde_json::to_string(&RequestPacket::Request { node_id, payload })? + }; + + self.driver + .notify(&chan, &payload) + .await + .map_err(Error::Driver)?; + Ok(()) + } +} + fn hash_chan(chan: &str) -> String { let hash = xxhash_rust::xxh3::xxh3_64(chan.as_bytes()); format!("ch_{:x}", hash)