weavegraph 0.3.0

Graph-driven, concurrent agent workflow framework with versioned state, deterministic barrier merges, and rich diagnostics.
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

//! Demonstrates event bus backpressure handling and lag recovery.
//!
//! This example shows:
//! - How to detect lagged event streams
//! - Metrics for monitoring dropped events
//! - Proper handling of RecvError::Lagged

use std::time::Duration;
use tokio::time::sleep;
use weavegraph::event_bus::Event;
use weavegraph::runtimes::{EventBusConfig, SinkConfig};

#[tokio::main]
async fn main() {
    // Create a bus with tiny capacity to trigger lag
    let bus = EventBusConfig::new(2, vec![SinkConfig::Memory]).build_event_bus();
    let emitter = bus.get_emitter();
    let mut stream = bus.subscribe();

    // Flood the bus
    for i in 0..100 {
        emitter
            .emit(Event::diagnostic("flood", format!("msg {i}")))
            .ok();
    }

    // Attempt to consume - will get Lagged error
    sleep(Duration::from_millis(10)).await;

    // Drain for a short period and then exit. This prevents the example from hanging.
    // `next_timeout` retries on lag and returns None on timeout or channel close.
    while let Some(event) = stream.next_timeout(Duration::from_millis(50)).await {
        println!("Received: {}", event.message());
    }

    let metrics = bus.metrics();
    println!("Capacity: {}", metrics.capacity);
    println!("Total dropped: {}", metrics.dropped);
}