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
//! 0.21 F: `monitor.subscribe::<P>()` streaming consumer demo.
//!
//! Subscribes to HTTP messages parsed off the monitor's run loop
//! and prints them from a separate tokio task. Both the
//! subscriber AND any registered `.on::<Http>` handler see every
//! message — the broadcast slot fans out per subscriber clone.
//!
//! Run:
//!
//! ```sh
//! cargo run --example monitor_stream_consumer \
//! --features "tokio,flow,http" -- eth0
//! ```
//!
//! Then in another shell: `curl http://example.com` or similar
//! over the watched interface.
use std::time::Duration;
use netring::monitor::Monitor;
use netring::protocol::builtin::Http;
// 0.21 H.2: `Monitor` is `Send` since flowscope 0.13's typed
// driver became `Send + Sync`. We can use the default
// multi-thread runtime and the regular `tokio::spawn`.
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let iface = std::env::args().nth(1).unwrap_or_else(|| "lo".into());
let monitor = Monitor::builder()
.interface(&iface)
// `with_broadcast::<Http>` registers Http as a broadcast
// slot — exactly the spot where `subscribe::<Http>()`
// pulls subscriber clones from. Calling `.protocol::<Http>()`
// instead would install a regular slot and `subscribe`
// would error with `BuildError::ProtocolNotBroadcast`.
.with_broadcast::<Http>()
.name("stream-consumer-demo")
.build()?;
// Mint a subscriber BEFORE the monitor moves into `run_for`.
let mut http_stream = monitor.subscribe::<Http>()?;
eprintln!(
"[stream-consumer] watching {iface} for 60s; \
HTTP messages will print as `[stream] <msg>` from the consumer task"
);
// Consumer task — independent of the run loop. Polls every
// 100ms; bounded drain via `recv_many(buf, 32)` prevents
// queue growth from monopolizing one tick.
let consumer = tokio::spawn(async move {
let mut buf = Vec::new();
let mut total = 0u64;
loop {
tokio::time::sleep(Duration::from_millis(100)).await;
let n = http_stream.recv_many(&mut buf, 32);
for msg in buf.drain(..) {
total += 1;
println!("[stream] http #{total}: {msg:?}");
}
if n == 0 && http_stream.subscribers() == 1 {
// Only the dispatcher's clone left → monitor is
// gone; consumer can exit.
break;
}
}
eprintln!("[stream-consumer] consumer drained {total} messages");
});
monitor.run_for(Duration::from_secs(60)).await?;
// Wait briefly for the consumer to drain any final messages.
let _ = tokio::time::timeout(Duration::from_secs(2), consumer).await;
Ok(())
}