Have an application-global state that's accessible from request handlers

[Pierstoval]

Hello everyone, for my new project (https://github.com/Orbitale/pagoo), I'd like my HTTP server to be run conjointedly with a "worker" system (see the app graph, I hope it helps).

I run the HTTP server, this works perfectly fine, but now I'd like my single request handler to be able to append elements to a queue, and alongside that, I'd like a separate thread to run an infinite loop that'd poll/listen (whatever method suits me as long as it works 🤣) to this queue and shift elements from the top of it on each iteration, one by one, and do something with it.

All I need is as simple as that, however, the not-so-simple part is the Queue itself: how to make sure Actix-based request handlers can "mutate" some data that's on the main thread while another specific thread will "retrieve" this data and do something with it?

I tried using a global "Queue" object, however I saw that Actix will create different instances of the App object for every spawned worker, so this means that every data injected into it via .app_data(...) will be Copy-ed, hence making it impossible to mutate from all threads.
I also tried using a channel from the mpsc tool, however I would actually need to do the opposite of all examples provided in the docs, and send the Sender object to all Actix workers so that I can use the Receiver object in one single separate thread in an infinite loop. The infinite loop & sending data from main thread do work, but the Sender object would also need to implement Send to be sent to all Actix workers, which is not the case.

I'm out of ideas now, and after hours of research, I'm starting to think about using an external child process and communicate via RPC or even a data source (like files), or even try to use in-memory Sqlite for that (by using your "Databases" guide), but that would be intense polling on a third-party library that I hope could be avoided somehow.

Does anyone have an idea on how to implement this kind of feature?

[asonix]

Are you looking for something like this?

#[derive(Debug)]
struct Message;
impl Message {
    fn new() -> Self {
        Self
    }
}

#[actix_web::get("/")]
async fn handler(
    sender: actix_web::web::Data<tokio::sync::mpsc::Sender<Message>>,
) -> actix_web::HttpResponse {
    let _ = sender.send(Message::new()).await;
    actix_web::HttpResponse::Ok().finish()
}

fn start_workers(mut receiver: tokio::sync::mpsc::Receiver<Message>) {
    tokio::spawn(async move {
        while let Some(msg) = receiver.recv().await {
            println!("Got message: {:?}", msg);
        }
    });
}

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

    start_workers(rx);

    actix_web::HttpServer::new(move || {
        actix_web::App::new()
            .app_data(actix_web::web::Data::new(tx.clone()))
            .service(handler)
    })
    .bind("0.0.0.0:8080")?
    .run()
    .await
}

[Pierstoval]

Will try this right now and check if it works! Thanks a lot for your suggestion!

[Pierstoval]

Okay, it works like a charm! I did not investigate further in the different types I can send, but for now I just send strings (I don't need more) and it's fine!

Only remaining thing for me is how to create tests for this part, but it's out of this ticket's scope 😉