rust async example

async2.rs

use std::sync::Arc;
use tokio::sync::{broadcast, Mutex};
use tokio::time::{sleep, Duration};

#[derive(Debug)]
struct SharedContext {
    counter: i32,
    last_updated_by: String,
}

// Worker that runs indefinitely
async fn worker(
    id: i32,
    context: Arc<Mutex<SharedContext>>,
    mut shutdown: broadcast::Receiver<()>,
) {
    loop {
        // Check if we received shutdown signal
        if shutdown.try_recv().is_ok() {
            println!("Worker {} received shutdown signal, stopping...", id);
            break;
        }

        // Simulate some async work
        sleep(Duration::from_secs(1)).await;

        // Update shared context
        let mut locked_context = context.lock().await;
        locked_context.counter += 1;
        locked_context.last_updated_by = format!("Worker {}", id);

        println!(
            "Worker {} processed item. Counter now at {}",
            id, locked_context.counter
        );
    }
}

// Monitor that periodically prints stats
async fn monitor(context: Arc<Mutex<SharedContext>>, mut shutdown: broadcast::Receiver<()>) {
    loop {
        if shutdown.try_recv().is_ok() {
            break;
        }

        let locked_context = context.lock().await;
        println!(
            "\n=== Status Report ===\nTotal processed: {}\nLast worker: {}\n",
            locked_context.counter, locked_context.last_updated_by
        );

        // Report every 5 seconds
        sleep(Duration::from_secs(5)).await;
    }
}

#[tokio::main]
async fn main() {
    // Create shared context
    let shared_context = Arc::new(Mutex::new(SharedContext {
        counter: 0,
        last_updated_by: String::new(),
    }));

    // Create shutdown channel
    let (shutdown_tx, _) = broadcast::channel(1);

    // Spawn workers
    let mut handles = vec![];

    for worker_id in 1..=3 {
        let context_clone = Arc::clone(&shared_context);
        let shutdown_rx = shutdown_tx.subscribe();

        let handle = tokio::spawn(async move {
            worker(worker_id, context_clone, shutdown_rx).await;
        });

        handles.push(handle);
    }

    // Spawn monitor
    let monitor_handle = tokio::spawn(monitor(
        Arc::clone(&shared_context),
        shutdown_tx.subscribe(),
    ));

    // Let the system run for a while
    println!("System started. Press Ctrl+C to stop...");

    // Wait for Ctrl+C
    tokio::signal::ctrl_c()
        .await
        .expect("Failed to listen for Ctrl+C");

    // Send shutdown signal
    println!("\nShutdown signal received, stopping workers...");
    shutdown_tx
        .send(())
        .expect("Failed to send shutdown signal");

    // Wait for all workers to complete
    for handle in handles {
        handle.await.unwrap();
    }

    // Wait for monitor to complete
    monitor_handle.await.unwrap();

    // Print final state
    let final_context = shared_context.lock().await;
    println!("\nFinal Statistics:");
    println!("Total items processed: {}", final_context.counter);
    println!("Last worker active: {}", final_context.last_updated_by);
}