Exposing data coming from a tokio::sync channel to a websocket.

[Krahos]

Hello everyone, I have a thread generating some data and sending it through a channel, I can't figure out how to, then, expose it to a websocket for clients to listen to. I'm aware of #2652 but it's still not clear.

The code I wrote so far is here.
My logic is:

1. Create the channel
2. Spawn the generator thread and pass it the sender
3. Spawn the server and use the receiver as application data
4. After the websocket handshake start listening to the receiver and send the data to the client
   Point 4 is what I'm struggling with but there could be mistakes elsewhere in this logic.

Do you have any pointers on how to solve this?

[Krahos]

Solved: I had to implement a different StreamHandler for the type I'm generating in the server. Full example code for anyone else stumbling on this:

use std::{net::TcpListener, thread, time::Duration};

use actix::{Actor, AsyncContext, SpawnHandle, StreamHandler};
use actix_web::{
    web::{self, Data, Payload},
    App, Error, HttpRequest, HttpResponse, HttpServer,
};
use actix_web_actors::ws::{self};

use tokio::sync::watch::{channel, Receiver};

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    // Define communication channel.
    let (sender, receiver) = channel(Message { int: 0 });

    // Spawn sender thread.
    let _handle = thread::spawn(move || loop {
        thread::sleep(Duration::from_secs(2));
        println!("Generated a message");
        sender
            .send(Message { int: 3 })
            .expect("Failed to write to channel");
        println!("Sent a message");
    });

    // Define server.
    let listener = TcpListener::bind("127.0.0.1:5568")?;

    let app_state = Data::new(AppState { receiver });
    let _server = HttpServer::new(move || {
        App::new()
            .route("/", web::get().to(index))
            .app_data(app_state.clone())
    })
    .listen(listener)?
    .run()
    .await;

    Ok(())
}

#[derive(Clone)]
struct AppState {
    receiver: Receiver<Message>,
}

#[derive(Debug)]
pub struct Message {
    pub int: i8,
}

impl ToString for Message {
    fn to_string(&self) -> String {
        self.int.to_string()
    }
}

struct MyWs {
    receiver: Receiver<Message>,
    spawn_handle: Option<SpawnHandle>,
}

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

    fn started(&mut self, ctx: &mut Self::Context) {
        let mut receiver = self.receiver.clone();
        self.spawn_handle = Some(ctx.add_stream(async_stream::stream! {
            while receiver.changed().await.is_ok() {
                yield receiver.borrow().to_string()
            };
        }));
    }
}

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

async fn index(
    req: HttpRequest,
    stream: Payload,
    app_state: Data<AppState>,
) -> Result<HttpResponse, Error> {
    ws::start(
        MyWs {
            receiver: app_state.receiver.clone(),
            spawn_handle: None,
        },
        &req,
        stream,
    )
}

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

[palinko91]

Thank you for sharing this it helped a lot with my own implementation. I've struggled for 1+ day but thanks to this now my server runs as I want :-) Now time to add authentication and the other ideas I want hopefully that will be easier with the new knowledge :-)