diff --git a/examples/benchmark.rs b/examples/benchmark.rs index 6a7b80d0c..37358b067 100644 --- a/examples/benchmark.rs +++ b/examples/benchmark.rs @@ -5,7 +5,7 @@ use std::sync::Arc; use wacore::net::{HttpClient, HttpRequest}; use wacore::proto_helpers::MessageExt; use wacore::store::InMemoryBackend; -use wacore::types::events::Event; +use wacore::types::events::{Event, EventKind}; use waproto::whatsapp as wa; use whatsapp_rust::TokioRuntime; use whatsapp_rust::bot::{Bot, MessageContext}; @@ -77,71 +77,79 @@ fn main() { .with_runtime(TokioRuntime); let mut bot = builder - .on_event(move |event, client| { - let admin_scan_url = admin_scan_url.clone(); - async move { - match &*event { - Event::Message(msg, info) => { - if let Some(text) = msg.text_content() - && text == "ping" - { - let ctx = MessageContext::from_parts(msg, info, client); - info!("Received text ping, sending pong..."); + .on_event_for( + &[ + EventKind::Message, + EventKind::PairingQrCode, + EventKind::Connected, + EventKind::LoggedOut, + ], + move |event, client| { + let admin_scan_url = admin_scan_url.clone(); + async move { + match &*event { + Event::Message(msg, info) => { + if let Some(text) = msg.text_content() + && text == "ping" + { + let ctx = MessageContext::from_parts(msg, info, client); + info!("Received text ping, sending pong..."); - let pong_text = format!("pong {}", ctx.info.id); - let reply_message = wa::Message { - conversation: Some(pong_text), - ..Default::default() - }; + let pong_text = format!("pong {}", ctx.info.id); + let reply_message = wa::Message { + conversation: Some(pong_text), + ..Default::default() + }; - if let Err(e) = ctx.send_message(reply_message).await { - error!("Failed to send pong reply: {}", e); + if let Err(e) = ctx.send_message(reply_message).await { + error!("Failed to send pong reply: {}", e); + } } } - } - Event::PairingQrCode { code, .. } => { - // Mirrors tests/e2e/src/lib.rs::spawn_qr_autoresponder_http. - // Auto-pair against the mock server's admin endpoint - // when the configured WS URL looks like a mock - // server; real WhatsApp connections fall back to - // manual scan via the printed code below. - if let Some(url) = admin_scan_url.as_ref() { - let http = UreqHttpClient::new(); - let req = HttpRequest { - url: url.clone(), - method: "POST".into(), - headers: HashMap::new(), - body: Some(code.as_bytes().to_vec()), - }; - match http.execute(req).await { - Ok(resp) if (200..300).contains(&resp.status_code) => { - info!("Auto-paired with mock server via {url}"); - } - Ok(resp) => { - warn!( - "mock admin POST returned status {}: {}", - resp.status_code, - String::from_utf8_lossy(&resp.body) - ); - } - Err(e) => { - warn!("mock admin POST transport error: {e}"); + Event::PairingQrCode { code, .. } => { + // Mirrors tests/e2e/src/lib.rs::spawn_qr_autoresponder_http. + // Auto-pair against the mock server's admin endpoint + // when the configured WS URL looks like a mock + // server; real WhatsApp connections fall back to + // manual scan via the printed code below. + if let Some(url) = admin_scan_url.as_ref() { + let http = UreqHttpClient::new(); + let req = HttpRequest { + url: url.clone(), + method: "POST".into(), + headers: HashMap::new(), + body: Some(code.as_bytes().to_vec()), + }; + match http.execute(req).await { + Ok(resp) if (200..300).contains(&resp.status_code) => { + info!("Auto-paired with mock server via {url}"); + } + Ok(resp) => { + warn!( + "mock admin POST returned status {}: {}", + resp.status_code, + String::from_utf8_lossy(&resp.body) + ); + } + Err(e) => { + warn!("mock admin POST transport error: {e}"); + } } + } else { + info!("Scan this QR code with WhatsApp:\n{code}"); } - } else { - info!("Scan this QR code with WhatsApp:\n{code}"); } + Event::Connected(_) => { + info!("✅ Bot connected successfully!"); + } + Event::LoggedOut(_) => { + error!("❌ Bot was logged out!"); + } + _ => {} } - Event::Connected(_) => { - info!("✅ Bot connected successfully!"); - } - Event::LoggedOut(_) => { - error!("❌ Bot was logged out!"); - } - _ => {} } - } - }) + }, + ) .build() .await .expect("Failed to build bot"); diff --git a/src/bot.rs b/src/bot.rs index c6094e002..a0d231bd6 100644 --- a/src/bot.rs +++ b/src/bot.rs @@ -5,7 +5,7 @@ use crate::store::commands::DeviceCommand; use crate::store::persistence_manager::PersistenceManager; use crate::store::traits::Backend; use crate::types::enc_handler::EncHandler; -use crate::types::events::{Event, EventHandler}; +use crate::types::events::{Event, EventHandler, EventInterest, EventKind}; use crate::types::message::MessageInfo; use anyhow::Result; use log::{info, warn}; @@ -119,15 +119,23 @@ impl MessageContext { type EventHandlerCallback = Arc, Arc) -> Pin + Send>> + Send + Sync>; +/// The user callback bundled with the set of event kinds it wants. Carrying the +/// interest here lets the bus skip materializing (and boxing) events the +/// callback ignores. +struct RegisteredHandler { + callback: EventHandlerCallback, + interest: EventInterest, +} + struct BotEventHandler { client: Arc, - event_handler: Option, + event_handler: Option, } impl EventHandler for BotEventHandler { fn handle_event(&self, event: Arc) { if let Some(handler) = &self.event_handler { - let handler_clone = handler.clone(); + let handler_clone = handler.callback.clone(); let client_clone = self.client.clone(); self.client @@ -138,6 +146,13 @@ impl EventHandler for BotEventHandler { .detach(); } } + + fn interest(&self) -> EventInterest { + self.event_handler + .as_ref() + .map(|h| h.interest) + .unwrap_or(EventInterest::none()) + } } /// Handle returned by [`Bot::run`] that can be awaited to wait for the @@ -168,7 +183,7 @@ impl std::future::Future for BotHandle { pub struct Bot { client: Arc, sync_task_receiver: Option>, - event_handler: Option, + event_handler: Option, pair_code_options: Option, } @@ -274,7 +289,7 @@ pub struct BotBuilder { http_client: Option>, runtime: Option>, // Optional fields - event_handler: Option, + event_handler: Option, custom_enc_handlers: HashMap>, override_version: Option<(u32, u32, u32)>, device_props_override: Option, @@ -444,14 +459,35 @@ impl BotBuilder { // ── Optional-field setters (available in any state) ────────────────────── impl BotBuilder { - pub fn on_event(mut self, handler: F) -> Self + /// Register a handler that receives every event kind. + pub fn on_event(self, handler: F) -> Self where F: Fn(Arc, Arc) -> Fut + Send + Sync + 'static, Fut: Future + Send + 'static, { - self.event_handler = Some(Arc::new(move |event, client| { - Box::pin(handler(event, client)) - })); + self.register_event_handler(EventInterest::ALL, handler) + } + + /// Register a handler that receives only the given event kinds. The bus + /// skips materializing (and boxing the handler future for) every other + /// kind, so a narrowly-scoped bot does not pay for events it ignores. + pub fn on_event_for(self, kinds: &[EventKind], handler: F) -> Self + where + F: Fn(Arc, Arc) -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, + { + self.register_event_handler(EventInterest::of(kinds), handler) + } + + fn register_event_handler(mut self, interest: EventInterest, handler: F) -> Self + where + F: Fn(Arc, Arc) -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, + { + self.event_handler = Some(RegisteredHandler { + callback: Arc::new(move |event, client| Box::pin(handler(event, client))), + interest, + }); self } diff --git a/src/history_sync.rs b/src/history_sync.rs index f7a2862e8..da940c22b 100644 --- a/src/history_sync.rs +++ b/src/history_sync.rs @@ -1,4 +1,4 @@ -use crate::types::events::{Event, LazyHistorySync}; +use crate::types::events::{Event, EventKind, LazyHistorySync}; use std::sync::Arc; use wacore::history_sync::{HistoryMsgSecretRecord, TcTokenCandidate, process_history_sync}; use wacore::store::traits::{MsgSecretEntry, TcTokenEntry}; @@ -145,8 +145,11 @@ impl Client { device_snapshot.pn.as_ref().map(|j| j.to_non_ad().user) }; - let has_listeners = self.core.event_bus.has_handlers(); - let retain_history_blob = has_listeners; + // Retain (and fully decompress) the blob only when a handler actually + // wants HistorySync. A message-only bot leaves this false, so the + // streaming decompress-and-parse path runs instead of materializing the + // whole payload just to drop it at dispatch. + let retain_history_blob = self.core.event_bus.has_handler_for(EventKind::HistorySync); // Small blobs (PushName, Recent): decode inline to avoid spawn_blocking overhead. // Large blobs: use blocking thread to avoid stalling the async runtime. diff --git a/src/main.rs b/src/main.rs index 03a71043e..87d9ddf78 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,7 +1,7 @@ use log::{error, info}; use std::sync::Arc; use wacore::proto_helpers::MessageExt; -use wacore::types::events::Event; +use wacore::types::events::{Event, EventKind}; use waproto::whatsapp as wa; use whatsapp_rust::TokioRuntime; use whatsapp_rust::bot::{Bot, MessageContext}; @@ -80,44 +80,53 @@ fn main() { } let mut bot = builder - .on_event(move |event, client| async move { - match &*event { - Event::PairingQrCode { code, timeout } => { - info!("----------------------------------------"); - info!( - "QR code received (valid for {} seconds):", - timeout.as_secs() - ); - info!("\n{}\n", code); - info!("----------------------------------------"); - } - Event::PairingCode { code, timeout } => { - info!("========================================"); - info!("PAIR CODE (valid for {} seconds):", timeout.as_secs()); - info!("Enter this code on your phone:"); - info!("WhatsApp > Linked Devices > Link a Device"); - info!("> Link with phone number instead"); - info!(""); - info!(" >>> {} <<<", code); - info!(""); - info!("========================================"); - } - Event::Message(msg, info) => { - let ctx = MessageContext::from_parts(msg, info, client); - if let Some(reply) = build_media_pong(msg) { - info!("Received media ping from {}", ctx.info.source.sender); - if let Err(e) = ctx.send_message(reply).await { - error!("Failed to send media pong: {}", e); + .on_event_for( + &[ + EventKind::PairingQrCode, + EventKind::PairingCode, + EventKind::Message, + EventKind::Connected, + EventKind::LoggedOut, + ], + move |event, client| async move { + match &*event { + Event::PairingQrCode { code, timeout } => { + info!("----------------------------------------"); + info!( + "QR code received (valid for {} seconds):", + timeout.as_secs() + ); + info!("\n{}\n", code); + info!("----------------------------------------"); + } + Event::PairingCode { code, timeout } => { + info!("========================================"); + info!("PAIR CODE (valid for {} seconds):", timeout.as_secs()); + info!("Enter this code on your phone:"); + info!("WhatsApp > Linked Devices > Link a Device"); + info!("> Link with phone number instead"); + info!(""); + info!(" >>> {} <<<", code); + info!(""); + info!("========================================"); + } + Event::Message(msg, info) => { + let ctx = MessageContext::from_parts(msg, info, client); + if let Some(reply) = build_media_pong(msg) { + info!("Received media ping from {}", ctx.info.source.sender); + if let Err(e) = ctx.send_message(reply).await { + error!("Failed to send media pong: {}", e); + } + } else if msg.text_content() == Some(PING_TRIGGER) { + handle_text_ping(&ctx).await; } - } else if msg.text_content() == Some(PING_TRIGGER) { - handle_text_ping(&ctx).await; } + Event::Connected(_) => info!("✅ Bot connected successfully!"), + Event::LoggedOut(_) => error!("❌ Bot was logged out!"), + _ => {} } - Event::Connected(_) => info!("✅ Bot connected successfully!"), - Event::LoggedOut(_) => error!("❌ Bot was logged out!"), - _ => {} - } - }) + }, + ) .build() .await .expect("Failed to build bot"); diff --git a/wacore/src/types/events.rs b/wacore/src/types/events.rs index 5954f3e3f..3207c9ede 100755 --- a/wacore/src/types/events.rs +++ b/wacore/src/types/events.rs @@ -192,8 +192,108 @@ impl Serialize for LazyHistorySync { } } +/// Discriminant for each [`Event`] variant, used to express handler interest +/// without materializing the event. One per `Event` variant, in declaration +/// order; the value doubles as a bit index in [`EventInterest`], so there can +/// be at most 64 kinds. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub enum EventKind { + Connected, + Disconnected, + PairSuccess, + PairError, + LoggedOut, + PairingQrCode, + PairingCode, + QrScannedWithoutMultidevice, + ClientOutdated, + Message, + Receipt, + UndecryptableMessage, + Notification, + ChatPresence, + Presence, + PictureUpdate, + UserAboutUpdate, + ContactUpdated, + ContactNumberChanged, + ContactSyncRequested, + GroupUpdate, + ContactUpdate, + IncomingCall, + PushNameUpdate, + SelfPushNameUpdated, + PinUpdate, + MuteUpdate, + ArchiveUpdate, + StarUpdate, + MarkChatAsReadUpdate, + DeleteChatUpdate, + DeleteMessageForMeUpdate, + HistorySync, + OfflineSyncPreview, + OfflineSyncCompleted, + DeviceListUpdate, + IdentityChange, + BusinessStatusUpdate, + StreamReplaced, + TemporaryBan, + ConnectFailure, + StreamError, + DisappearingModeChanged, + NewsletterLiveUpdate, + RawNode, + MexNotification, +} + +/// A set of [`EventKind`]s a handler wants delivered. The event bus skips +/// materializing and dispatching events whose kind no handler wants, so a +/// handler that subscribes to a few kinds never pays for boxing the others. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct EventInterest(u64); + +impl EventInterest { + /// Every kind. Default for handlers that don't narrow their interest. + pub const ALL: EventInterest = EventInterest(u64::MAX); + + /// No kinds. + pub const fn none() -> Self { + EventInterest(0) + } + + /// Interest in exactly the given kinds. + pub fn of(kinds: &[EventKind]) -> Self { + let mut bits = 0u64; + let mut i = 0; + while i < kinds.len() { + bits |= 1u64 << (kinds[i] as u8); + i += 1; + } + EventInterest(bits) + } + + /// Add a kind to the set. + pub const fn with(self, kind: EventKind) -> Self { + EventInterest(self.0 | (1u64 << (kind as u8))) + } + + /// Whether `kind` is in the set. + #[inline] + pub const fn wants(self, kind: EventKind) -> bool { + self.0 & (1u64 << (kind as u8)) != 0 + } +} + pub trait EventHandler: crate::sync_marker::MaybeSendSync { fn handle_event(&self, event: Arc); + + /// Which event kinds this handler wants. Defaults to all kinds, so the bus + /// keeps delivering everything to handlers that don't opt into a narrower + /// set. Override to let the bus skip materializing unwanted events. + fn interest(&self) -> EventInterest { + EventInterest::ALL + } } /// Event handler that forwards events to an async channel. @@ -250,6 +350,17 @@ impl CoreEventBus { .is_empty() } + /// Whether any registered handler is interested in `kind`. Lets callers + /// skip producing an event nobody would receive (e.g. retaining a large + /// `HistorySync` blob when only message-only handlers are registered). + pub fn has_handler_for(&self, kind: EventKind) -> bool { + self.handlers + .read() + .expect("RwLock should not be poisoned") + .iter() + .any(|h| h.interest().wants(kind)) + } + pub fn dispatch(&self, event: Event) { let handlers = self .handlers @@ -259,9 +370,18 @@ impl CoreEventBus { if handlers.is_empty() { return; } + // Skip materializing the event (Arc) and invoking handlers whose + // declared interest excludes this kind. A handler that subscribed to a + // few kinds never pays for boxing the events it ignores. + let kind = event.kind(); + if !handlers.iter().any(|h| h.interest().wants(kind)) { + return; + } let event = Arc::new(event); for handler in &handlers { - handler.handle_event(Arc::clone(&event)); + if handler.interest().wants(kind) { + handler.handle_event(Arc::clone(&event)); + } } } } @@ -529,6 +649,59 @@ pub struct MexNotification { } impl Event { + /// The [`EventKind`] discriminant for this event, used by the bus to test + /// handler interest before materializing the event. + pub fn kind(&self) -> EventKind { + match self { + Event::Connected(_) => EventKind::Connected, + Event::Disconnected(_) => EventKind::Disconnected, + Event::PairSuccess(_) => EventKind::PairSuccess, + Event::PairError(_) => EventKind::PairError, + Event::LoggedOut(_) => EventKind::LoggedOut, + Event::PairingQrCode { .. } => EventKind::PairingQrCode, + Event::PairingCode { .. } => EventKind::PairingCode, + Event::QrScannedWithoutMultidevice(_) => EventKind::QrScannedWithoutMultidevice, + Event::ClientOutdated(_) => EventKind::ClientOutdated, + Event::Message(_, _) => EventKind::Message, + Event::Receipt(_) => EventKind::Receipt, + Event::UndecryptableMessage(_) => EventKind::UndecryptableMessage, + Event::Notification(_) => EventKind::Notification, + Event::ChatPresence(_) => EventKind::ChatPresence, + Event::Presence(_) => EventKind::Presence, + Event::PictureUpdate(_) => EventKind::PictureUpdate, + Event::UserAboutUpdate(_) => EventKind::UserAboutUpdate, + Event::ContactUpdated(_) => EventKind::ContactUpdated, + Event::ContactNumberChanged(_) => EventKind::ContactNumberChanged, + Event::ContactSyncRequested(_) => EventKind::ContactSyncRequested, + Event::GroupUpdate(_) => EventKind::GroupUpdate, + Event::ContactUpdate(_) => EventKind::ContactUpdate, + Event::IncomingCall(_) => EventKind::IncomingCall, + Event::PushNameUpdate(_) => EventKind::PushNameUpdate, + Event::SelfPushNameUpdated(_) => EventKind::SelfPushNameUpdated, + Event::PinUpdate(_) => EventKind::PinUpdate, + Event::MuteUpdate(_) => EventKind::MuteUpdate, + Event::ArchiveUpdate(_) => EventKind::ArchiveUpdate, + Event::StarUpdate(_) => EventKind::StarUpdate, + Event::MarkChatAsReadUpdate(_) => EventKind::MarkChatAsReadUpdate, + Event::DeleteChatUpdate(_) => EventKind::DeleteChatUpdate, + Event::DeleteMessageForMeUpdate(_) => EventKind::DeleteMessageForMeUpdate, + Event::HistorySync(_) => EventKind::HistorySync, + Event::OfflineSyncPreview(_) => EventKind::OfflineSyncPreview, + Event::OfflineSyncCompleted(_) => EventKind::OfflineSyncCompleted, + Event::DeviceListUpdate(_) => EventKind::DeviceListUpdate, + Event::IdentityChange(_) => EventKind::IdentityChange, + Event::BusinessStatusUpdate(_) => EventKind::BusinessStatusUpdate, + Event::StreamReplaced(_) => EventKind::StreamReplaced, + Event::TemporaryBan(_) => EventKind::TemporaryBan, + Event::ConnectFailure(_) => EventKind::ConnectFailure, + Event::StreamError(_) => EventKind::StreamError, + Event::DisappearingModeChanged(_) => EventKind::DisappearingModeChanged, + Event::NewsletterLiveUpdate(_) => EventKind::NewsletterLiveUpdate, + Event::RawNode(_) => EventKind::RawNode, + Event::MexNotification(_) => EventKind::MexNotification, + } + } + pub fn as_message(&self) -> Option<(&Arc, &MessageInfo)> { if let Event::Message(msg, info) = self { Some((msg, &**info)) @@ -1207,4 +1380,58 @@ mod tests { ); assert!(!ConnectFailureReason::from(499).is_logged_out()); } + + #[test] + fn interest_filters_dispatch() { + use std::sync::Mutex; + use std::sync::atomic::{AtomicUsize, Ordering}; + + struct Recorder { + kinds: Mutex>, + interest: EventInterest, + } + impl EventHandler for Recorder { + fn handle_event(&self, event: Arc) { + self.kinds.lock().unwrap().push(event.kind()); + } + fn interest(&self) -> EventInterest { + self.interest + } + } + + let bus = CoreEventBus::new(); + let only_msg = Arc::new(Recorder { + kinds: Mutex::new(Vec::new()), + interest: EventInterest::of(&[EventKind::Message]), + }); + let all = Arc::new(Recorder { + kinds: Mutex::new(Vec::new()), + interest: EventInterest::ALL, + }); + bus.add_handler(only_msg.clone()); + bus.add_handler(all.clone()); + + bus.dispatch(Event::Connected(Connected)); + + // The narrow handler (Message-only) was skipped; the ALL handler got it. + assert!(only_msg.kinds.lock().unwrap().is_empty()); + assert_eq!(*all.kinds.lock().unwrap(), vec![EventKind::Connected]); + + // A kind nobody wants is dropped before materialization: prove the bus + // never invokes a handler for it. + static CALLS: AtomicUsize = AtomicUsize::new(0); + struct Counter; + impl EventHandler for Counter { + fn handle_event(&self, _: Arc) { + CALLS.fetch_add(1, Ordering::SeqCst); + } + fn interest(&self) -> EventInterest { + EventInterest::of(&[EventKind::Message]) + } + } + let bus2 = CoreEventBus::new(); + bus2.add_handler(Arc::new(Counter)); + bus2.dispatch(Event::Connected(Connected)); + assert_eq!(CALLS.load(Ordering::SeqCst), 0); + } }