async dispatch api

gistfile1.rs

use crate::CreateState::Dispatched;
use actix_web::{App};
use std::collections::HashMap;
use std::sync::mpsc::channel;
use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

use actix_web::{HttpResponse, Responder, get, web};
use serde::Serialize;

enum CreateState {
    Running(usize),
    Dispatched(usize),
}

#[derive(Debug)]
enum Command {
    Clean(Sender<CreateState>),
    Get((usize, Sender<String>)),
}

struct AppState {
    sender: Sender<Command>,
}

fn clean_room() {
    println!("Staring Cleaning Room");
    thread::sleep(Duration::from_secs(10));
    println!("Finished Cleaning Room");
}

fn do_clean_job(
    job_id: usize,
    rx_back: Sender<CreateState>,
    job_results_: Arc<Mutex<HashMap<usize, String>>>,
) {
    rx_back.send(Dispatched(job_id)).unwrap();
    job_results_
        .lock()
        .unwrap()
        .insert(job_id, format!("Job {} started!", job_id));

    clean_room();

    let mut results_map = job_results_.lock().unwrap();
    results_map.entry(job_id).and_modify(|existing_message| {
        existing_message.push_str("\n ->  completed!");
    });
}

fn job_manager(rx: Receiver<Command>) {
    let current_job_id = Arc::new(Mutex::new(0usize));

    let running = Arc::new(Mutex::new(false));
    let job_results: Arc<Mutex<HashMap<usize, String>>> = Arc::new(Mutex::new(HashMap::new()));

    loop {
        let command = rx.recv().unwrap();
        println!("Customer requested: {command:?}");

        match command {
            Command::Clean(rx_back) if *running.lock().unwrap() => {
                println!("Job manager is busy. Denying request.");
                let current_id = *current_job_id.lock().unwrap();
                rx_back
                    .send(CreateState::Running(current_id))
                    .expect("TODO: panic message");
                continue;
            }
            Command::Clean(rx_back) => {
                let running_ = Arc::clone(&running);
                let job_results_ = Arc::clone(&job_results);
                let current_job_id_ = Arc::clone(&current_job_id);

                thread::spawn(move || {
                    *running_.lock().unwrap() = true;

                    let job_id_for_this_task = {
                        let mut id_guard = current_job_id_.lock().unwrap();
                        let id = *id_guard;
                        *id_guard += 1; // Increment for the *next* job
                        id
                    };

                    do_clean_job(job_id_for_this_task, rx_back, job_results_);

                    println!("Finalized Customer Request");
                    *running_.lock().unwrap() = false;
                });
            }
            Command::Get((id, rx_back)) => {
                let job_results_guard = job_results.lock().unwrap();
                let job_result_string = job_results_guard
                    .get(&id)
                    .cloned() // This clones the String inside the Option<&String>
                    .unwrap_or_else(|| "None".to_string());
                let _ = rx_back.send(job_result_string);
            }
        }
    }
}

#[derive(Serialize)]
struct JobCreationResponse {
    status: String,
    message: String,
}

#[get("/")]
async fn new_job(data: web::Data<AppState>) -> impl Responder {
    let tx = data.sender.clone();

    let (tx_feedback, rx_feedback) = std::sync::mpsc::channel();
    tx.send(Command::Clean(tx_feedback))
        .expect("Failed to send request to manager");

    match rx_feedback.recv().unwrap() {
        CreateState::Running(id) => HttpResponse::Locked().json(JobCreationResponse {
            status: "locked".to_string(),
            message: format!("Respource is locked by job {id}"),
        }),
        Dispatched(id) => HttpResponse::Ok().json(JobCreationResponse {
            status: "job created".to_string(),
            message: format!("/jobs/{id}"),
        }),
    }
}

#[derive(Serialize)]
struct JobResponse {
    status: String,
    id: usize,
}

#[get("/jobs/{job_id}")]
async fn get_jobs(data: web::Data<AppState>, job_id: web::Path<usize>) -> impl Responder {
    let tx = data.sender.clone();

    let (tx_feedback, rx_feedback) = std::sync::mpsc::channel();
    tx.send(Command::Get((job_id.clone(), tx_feedback)))
        .expect("Failed to send request to manager");

    let result = rx_feedback.recv().unwrap();

    match result.as_str() {
        "None" => HttpResponse::NotFound().json(JobResponse {
            status: "could not find job".to_string(),
            id: job_id.into_inner(),
        }),
        result => HttpResponse::Ok().json(JobResponse {
            status: result.to_string(),
            id: job_id.into_inner(),
        }),
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let (tx, rx) = channel();

    thread::spawn(move || {
        job_manager(rx);
    });

    let app_data = web::Data::new(AppState { sender: tx.clone() });
    actix_web::HttpServer::new(move || {
        App::new()
            .app_data(app_data.clone())
            .service(get_jobs)
            .service(new_job)
    })
    .bind(("127.0.0.1", 8080))
    .unwrap()
    .run()
    .await
}