peace_effort.rs

peace_effort.rs

// Dependencies used: tokio, hyper, hyper-util, http-body-util, serde, serde_json, url
use http_body_util::Full;
use hyper::body::Bytes;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response, StatusCode};
use hyper_util::rt::TokioIo;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use tokio::net::TcpListener;
use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};
use url::Url;

#[derive(Debug, Clone)]
struct ControlMessage {
    target: String,
    count: i32,
}

struct AppContext {
    tx: mpsc::Sender<ControlMessage>,
    worker_active: Arc<AtomicBool>,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let (tx, mut rx) = mpsc::channel::<ControlMessage>(32);
    let worker_active = Arc::new(AtomicBool::new(false));
    let worker_active_worker_ref = Arc::clone(&worker_active);

    // 1. Background Worker
    tokio::spawn(async move {
        while let Some(msg) = rx.recv().await {
            // Check if already active (primitive guard)
            if !worker_active_worker_ref.load(Ordering::SeqCst) {
                worker_active_worker_ref.store(true, Ordering::SeqCst);
                
                do_stuff(msg).await;
                
                worker_active_worker_ref.store(false, Ordering::SeqCst);
            }
        }
    });

    let ctx = Arc::new(AppContext {
        tx,
        worker_active,
    });

    let addr = SocketAddr::from(([0, 0, 0, 0], 8080));
    let listener = TcpListener::bind(addr).await?;
    println!("Server running on http://{}", addr);

    // 2. Server Loop (Hyper 1.0 style)
    loop {
        let (stream, _) = listener.accept().await?;
        let io = TokioIo::new(stream);
        let ctx_clone = Arc::clone(&ctx);

        tokio::task::spawn(async move {
            if let Err(err) = http1::Builder::new()
                .serve_connection(io, service_fn(move |req| {
                    handle_request(req, Arc::clone(&ctx_clone))
                }))
                .await
            {
                eprintln!("Error serving connection: {:?}", err);
            }
        });
    }
}

// 3. Manual Router using Hyper
async fn handle_request(
    req: Request<hyper::body::Incoming>,
    ctx: Arc<AppContext>,
) -> Result<Response<Full<Bytes>>, hyper::Error> {
    let path = req.uri().path();
    
    match path {
        "/" => {
            // Manual Query Parsing using 'url' crate
            let query_str = req.uri().query().unwrap_or("");
            let dummy_url = Url::parse(&format!("http://localhost/?{}", query_str)).unwrap();
            let pairs = dummy_url.query_pairs();
            
            let mut target = String::new();
            let mut count = 0;

            for (key, value) in pairs {
                if key == "target" { target = value.clone().into_owned(); }
                if key == "count" { count = value.parse::<i32>().unwrap_or(0); }
            }

            if count <= 0 {
                return Ok(Response::builder()
                    .status(StatusCode::BAD_REQUEST)
                    .body(Full::new(Bytes::from("Invalid Count")))
                    .unwrap());
            }

            let _ = ctx.tx.send(ControlMessage { target: target.clone(), count }).await;

            let res = format!("Control message issued for {} to {}", count, target);
            Ok(Response::new(Full::new(Bytes::from(res))))
        }

        "/status" => {
            let status = if ctx.worker_active.load(Ordering::SeqCst) {
                "ACTIVE"
            } else {
                "INACTIVE"
            };
            Ok(Response::new(Full::new(Bytes::from(status))))
        }

        _ => {
            Ok(Response::builder()
                .status(StatusCode::NOT_FOUND)
                .body(Full::new(Bytes::from("Not Found")))
                .unwrap())
        }
    }
}

async fn do_stuff(msg: ControlMessage) {
    for i in 1..=msg.count {
        println!("[{}/{}] Spreading peace to {}...", i, msg.count, msg.target);
        sleep(Duration::from_millis(500)).await;
    }
}

https://play.rust-lang.org/?version=stable&mode=debug&edition=2024&gist=a8b2ec49197d55ec2fa79d9ba28e1f11