1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
//! Demonstrates `on_async` — async-handler I/O escape hatch.
//!
//! Two handlers cooperate via a shared `Arc<AtomicU64>`:
//!
//! - **sync `on::<FlowStarted<Tcp>>`** — bumps an in-memory
//! counter. Cost: zero allocations per dispatch.
//! - **async `on_async::<FlowStarted<Tcp>>`** — captures an
//! `Arc<SimulatedPool>` and pretends to do remote I/O. Each
//! event yields once and then increments the pool's counter.
//! Cost: one boxed future allocation per dispatch.
//!
//! Both handlers fire on every flow start. The sync one finishes
//! first; the async one's future is then awaited to completion
//! before the next event is processed.
//!
//! Run:
//!
//! ```sh
//! cargo run --example monitor_async_handler \
//! --features "tokio,flow" -- eth0
//! ```
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use netring::prelude::*;
/// Stand-in for a real I/O resource (Redis pool, HTTP client, …).
/// In a real monitor you'd hold an `Arc<bb8::Pool<…>>` or similar.
struct SimulatedPool {
publishes: AtomicU64,
}
impl SimulatedPool {
async fn publish(&self) -> Result<(), netring::Error> {
tokio::task::yield_now().await;
self.publishes.fetch_add(1, Ordering::Relaxed);
Ok(())
}
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let iface = std::env::args().nth(1).unwrap_or_else(|| "lo".into());
let dur_secs: u64 = std::env::args()
.nth(2)
.and_then(|s| s.parse().ok())
.unwrap_or(10);
eprintln!("monitor_async_handler: capturing on {iface} for {dur_secs}s");
let sync_count = Arc::new(AtomicU64::new(0));
let pool = Arc::new(SimulatedPool {
publishes: AtomicU64::new(0),
});
let sync_count_h = Arc::clone(&sync_count);
let pool_h = Arc::clone(&pool);
let monitor = Monitor::builder()
.interface(&iface)
.protocol::<Tcp>()
.on::<FlowStarted<Tcp>>(move |_evt: &FlowStarted<Tcp>| {
sync_count_h.fetch_add(1, Ordering::Relaxed);
Ok(())
})
.on_async::<FlowStarted<Tcp>, _>(move |_evt: &FlowStarted<Tcp>| {
let pool = Arc::clone(&pool_h);
async move { pool.publish().await }
})
.build()?;
monitor.run_for(Duration::from_secs(dur_secs)).await?;
eprintln!(
"sync handler fires: {}",
sync_count.load(Ordering::Relaxed)
);
eprintln!(
"async pool publishes: {}",
pool.publishes.load(Ordering::Relaxed)
);
Ok(())
}