Add awc websocket client to add_stream in actix

1
votes

I'm trying to implement a websocket client within the actix framework, but so far I'm having a hard time make this work.

I started with awc websocket client

    use actix::{Actor, AsyncContext, Context, Running, StreamHandler, WrapFuture};
use actix_web_actors::ws::{Frame, ProtocolError};
use awc::ws::{Codec, Message};
use awc::BoxedSocket;
use futures::{FutureExt, SinkExt, StreamExt};
use log::info;
use openssl::ssl::SslConnector;
use serde_json::json;
use tokio::net::TcpSocket;

pub struct RosClient {
    pub connection: Option<TcpSocket>,
}

impl RosClient {
    pub fn new() -> Self {
        Self { connection: None }
    }
}

impl Actor for RosClient {
    type Context = Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        let handle = async move {
            let ssl = {
                let mut ssl = SslConnector::builder(openssl::ssl::SslMethod::tls()).unwrap();
                let _ = ssl.set_alpn_protos(b"\x08http/1.1");
                ssl.build()
            };
            let connector = awc::Connector::new().ssl(ssl).finish();
            let ws = awc::ClientBuilder::new()
                .connector(connector)
                .finish()
                .ws("ws://127.0.0.1:9090")
                .set_header("Host", "0.0.0.0:9090");

            let (resp, mut connection) = ws
                .connect()
                .await
                .map_err(|e| {
                    println!("{:?}", e);
                    e
                })
                .unwrap();
            println!("{:?}", resp);
            let message = json!({
                "op": "subscribe",
                "topic": "/client_count"
            })
            .to_string();
            println!("Is write ready {}", connection.is_write_ready());
            let t = connection.send(Message::Text(message)).await.unwrap();
        };
        ctx.spawn(handle.into_actor(self));
        // ctx.add_stream(handle.into_stream());
    }

    fn stopping(&mut self, ctx: &mut Self::Context) -> Running {
        Running::Stop
    }
}

impl StreamHandler<Result<Frame, ProtocolError>> for RosClient {
    fn handle(&mut self, item: Result<Frame, ProtocolError>, ctx: &mut Self::Context) {
        match item.unwrap() {
            Frame::Text(text_bytes) => {
                println!("Text {:?}", std::str::from_utf8(&text_bytes));
            }
            Frame::Binary(_) => {}
            Frame::Continuation(_) => {}
            Frame::Ping(bin) => {
                println!("Ping {:?}", std::str::from_utf8(&bin))
            }
            Frame::Pong(_) => {}
            Frame::Close(_) => {}
        }
    }
}

But there are some things I'm still not able to do. The code above is obviously incomplete, but I felt that it would at least provide some starting grounds for the discussion.

  • First: How can I implement the add_stream properly?

Right now, I create the future, which allows me to consume from the connection or to send messages from the connection, but I need to add a stream to the context.

  • Second: The connection result is a Framed<BoxedSocket, Codec>. I need to keep a reference to the connection somewhere, so that I can send messages to the server. Right now, I can't just do self.connection = connection, because I can't clone it.

So, what would be the right way of implementing that?

Thank you, Fabricio

rustactix-webrust-actix