Skip to content

perf(events): typed event subscription to skip boxing unwanted events #676

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jlucaso1 merged 3 commits into from
Jun 1, 2026
Merged

perf(events): typed event subscription to skip boxing unwanted events #676

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b3725e6
perf(events): typed event subscription to skip boxing unwanted events
jlucaso1 Jun 1, 2026
93d1833
examples: use on_event_for in the main and benchmark examples
jlucaso1 Jun 1, 2026
a35149d
perf(history-sync): gate blob retention on typed HistorySync interest
jlucaso1 Jun 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
122 changes: 65 additions & 57 deletions examples/benchmark.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -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");
Expand Down
54 changes: 45 additions & 9 deletions src/bot.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -119,15 +119,23 @@ impl MessageContext {
type EventHandlerCallback =
Arc<dyn Fn(Arc<Event>, Arc<Client>) -> Pin<Box<dyn Future<Output = ()> + 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<Client>,
event_handler: Option<EventHandlerCallback>,
event_handler: Option<RegisteredHandler>,
}

impl EventHandler for BotEventHandler {
fn handle_event(&self, event: Arc<Event>) {
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
Expand All @@ -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
Expand Down Expand Up @@ -168,7 +183,7 @@ impl std::future::Future for BotHandle {
pub struct Bot {
client: Arc<Client>,
sync_task_receiver: Option<async_channel::Receiver<crate::sync_task::MajorSyncTask>>,
event_handler: Option<EventHandlerCallback>,
event_handler: Option<RegisteredHandler>,
pair_code_options: Option<PairCodeOptions>,
}

Expand Down Expand Up @@ -274,7 +289,7 @@ pub struct BotBuilder<B = Missing, T = Missing, H = Missing, R = Missing> {
http_client: Option<Arc<dyn crate::http::HttpClient>>,
runtime: Option<Arc<dyn Runtime>>,
// Optional fields
event_handler: Option<EventHandlerCallback>,
event_handler: Option<RegisteredHandler>,
custom_enc_handlers: HashMap<String, Arc<dyn EncHandler>>,
override_version: Option<(u32, u32, u32)>,
device_props_override: Option<DevicePropsOverride>,
Expand Down Expand Up @@ -444,14 +459,35 @@ impl<B, T, H> BotBuilder<B, T, H, Missing> {
// ── Optional-field setters (available in any state) ──────────────────────

impl<B, T, H, R> BotBuilder<B, T, H, R> {
pub fn on_event<F, Fut>(mut self, handler: F) -> Self
/// Register a handler that receives every event kind.
pub fn on_event<F, Fut>(self, handler: F) -> Self
where
F: Fn(Arc<Event>, Arc<Client>) -> Fut + Send + Sync + 'static,
Fut: Future<Output = ()> + 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<F, Fut>(self, kinds: &[EventKind], handler: F) -> Self
where
F: Fn(Arc<Event>, Arc<Client>) -> Fut + Send + Sync + 'static,
Fut: Future<Output = ()> + Send + 'static,
{
self.register_event_handler(EventInterest::of(kinds), handler)
}

fn register_event_handler<F, Fut>(mut self, interest: EventInterest, handler: F) -> Self
where
F: Fn(Arc<Event>, Arc<Client>) -> Fut + Send + Sync + 'static,
Fut: Future<Output = ()> + Send + 'static,
{
self.event_handler = Some(RegisteredHandler {
callback: Arc::new(move |event, client| Box::pin(handler(event, client))),
interest,
});
self
}

Expand Down
9 changes: 6 additions & 3 deletions src/history_sync.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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};
Expand Down Expand Up @@ -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.
Expand Down
81 changes: 45 additions & 36 deletions src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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};
Expand Down Expand Up @@ -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");
Expand Down
Loading
Loading