Use channel sender in service

[jb-alvarado]

Hello everybody,
I try to use a channel sender in a service. In my real program I would like to use it, for long running tasks. The idea is, that with a API endpoint I start a task and with websocket I get a status progress.

The problem I have is, that I don't get any message from the channel send command in a service. I also tried different kind of channels (from tokio and async-channel), but with no different.

Here is a minimal running example:

use std::{fmt, thread, time::Duration};

use actix::prelude::*;
use actix_web::{
    get, middleware, web, App, Error, HttpRequest, HttpResponse, HttpServer, Responder,
};
use actix_web_actors::ws;
use async_channel::{bounded, Receiver, Sender};

use serde::Serialize;

#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize)]
struct Task {
    name: String,
    message: String,
    current: usize,
}

impl Task {
    pub fn new(name: &str, message: &str, current: usize) -> Self {
        Self {
            name: name.to_string(),
            message: message.to_string(),
            current,
        }
    }
}

impl fmt::Display for Task {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "{{ \"name\": \"{}\", \"massage\": \"{}\", \"current\": {} }}",
            self.name, self.message, self.current
        )
    }
}

struct WebSocket {
    receiver: web::Data<Receiver<Task>>,
    spawn_handle: Option<SpawnHandle>,
}

impl WebSocket {
    fn new(receiver: web::Data<Receiver<Task>>) -> Self {
        Self {
            receiver,
            spawn_handle: None,
        }
    }
}

impl Actor for WebSocket {
    type Context = ws::WebsocketContext<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        println!("-- actor");
        let receiver = self.receiver.clone();
        self.spawn_handle = Some(ctx.add_stream(async_stream::stream! {
            while let Ok(message) = receiver.recv().await {
                println!("-- msg: {message}");
                yield message.to_string();
            }
        }));
    }
}

impl StreamHandler<String> for WebSocket {
    fn handle(&mut self, msg: String, ctx: &mut Self::Context) {
        ctx.text(msg);
    }
}

impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WebSocket {
    fn handle(&mut self, _msg: Result<ws::Message, ws::ProtocolError>, _ctx: &mut Self::Context) {
        println!("Received a message")
    }
}

async fn status_ws(
    req: HttpRequest,
    stream: web::Payload,
    recv: web::Data<Receiver<Task>>,
) -> Result<HttpResponse, Error> {
    println!("-- client connect");
    ws::start(WebSocket::new(recv), &req, stream)
}

#[get("/hello/")]
async fn say_hello(sender: web::Data<Sender<Task>>) -> impl Responder {
    let task = Task::new("echo", "hallo", 44);
    println!("say hello");
    sender
        .send(task.clone())
        .await
        .expect("Failed to write to channel");

    web::Json(task)
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(move || {
        let (sender, receiver) = bounded::<Task>(200);
        let sender2 = sender.clone();
        let sender = web::Data::new(sender);
        let receiver = web::Data::new(receiver);

        let mut count = 1;
        let _handle = thread::spawn(move || loop {
            thread::sleep(Duration::from_secs(2));
            sender2
                .send_blocking(Task::new("echo", "from loop", count))
                .expect("Failed to write to channel");

            count += 1;
        });

        App::new()
            .app_data(sender)
            .app_data(receiver)
            .wrap(middleware::Logger::default())
            .service(web::resource("/ws/status").route(web::get().to(status_ws)))
            .service(web::scope("/api").service(say_hello))
    })
    .bind(("127.0.0.1", 8077))?
    .run()
    .await
}

In my websocket client I get messages from thread _handle, but when I connect to /api/hello/ I don't get any message.

Can you give my a hint, what I need to do here?

[jb-alvarado]

I got now the solution. I created the channel in the wrong context. It needed to go at the begin of main function.