netring 0.24.0

High-performance zero-copy packet I/O for Linux (AF_PACKET TPACKET_V3 + AF_XDP)
Documentation 
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(())
}