Trying to use kameo with tokio-tungstentie

[bananasov]

Heyo! I've been trying kameo and so far I am absolutely loving the library, although i can't seem to wrap my brain around how to use the stream example with tokio-tungstenite websocket streams.

I'm trying to implement a client for a WebSocket system that sends messages to the client and receives them from a client, I want to be able to send the incoming messages from the stream to the person that created the actor and vice versa.

I've tried to just use tokio_tungstenite::connect_async return value with ActorRef::attach_stream but that just resulted in errors I don't understand.

let (ws_stream, _) = connect_async(&url).await.expect("Failed to connect");
let stream = Box::pin(ws_stream);
actor_ref.attach_stream(stream, "1st stream", "1st stream");

Does anyone have an idea? I am absolutely stumped.

[tqwewe]

Hi @bananasov, thanks for asking this! The errors returned are definitely quite obscure.

Essentially, in order to meet the requirements of attach_stream, your actor needs to implement Message<StreamMessage<T, S, F>>. StreamMessage is a type provided by kameo, and is essentially just a wrapper type around the stream items. S and F can be any type you want for sending through context when the stream starts and finishes.

Looking into the Stream impl of WebSocketStream, the Item type is Result<Message, Error>, which indicates your actor should have the following impl block:

use kameo::message::{Context, Message, StreamMessage};

impl
    Message<
        StreamMessage<
            Result<tungstenite::protocol::Message, tungstenite::error::Error>,
            &'static str,
            &'static str,
        >,
    > for MyActor
{
    type Reply = Result<(), tungstenite::error::Error>;

    async fn handle(&self, msg: StreamMessage, _ctx: Context<'_, Self, Self::Reply>) -> Self::Reply {
        match msg {
            StreamMessage::Next(ws_result) => {
                let ws_msg = ws_result?;
                // handle the websocket message
            }
            StreamMessage::Started(started_msg) => /* handle started */,
            StreamMessage::Finished(finished_msg) => /* handle finished */,
        }
    }
}

With this implementation, you should be able to call actor_ref.attach_stream using the stream, and your two &'static str items :)

Hopefully this clears it up and solves the errors you're having

[bananasov]

That clears it up by a lot! I'm wondering how the actor ref would receive messages from the stream though, without sending a message first

[tqwewe]

The attach stream method essentially spawns a dedicated tokio task which iterates the stream and forwards the items as messages to the actor.

kameo/src/actor/actor_ref.rs

Lines 521 to 549 in f2eaf24

	tokio::spawn(async move {
	actor_ref
	.tell(StreamMessage::Started(start_value))
	.send()
	.await?;
	loop {
	tokio::select! {
	msg = stream.next() => {
	match msg {
	Some(msg) => {
	actor_ref.tell(StreamMessage::Next(msg)).send().await?;
	}
	None => break,
	}
	}
	_ = actor_ref.wait_for_stop() => {
	return Ok(stream);
	}
	}
	}
	actor_ref
	.tell(StreamMessage::Finished(finish_value))
	.send()
	.await?;
	Ok(stream)
	})

This is quite useful if you want to attach the stream in the background. However, if you're already in a running task lives the duration of the websocket connection, you can also manually message your actor:

while let Some(item) = stream.next().await? {
    actor_ref.tell(item).send().await?;
}

If your actors entire purpose is just for handling the websocket connection and will not be messaged in any part of your system (i.e. your only message impl is for StreamMessage and nothing else), then you may not even need an actor for this and instead simply handle the websocket connection in a tokio task.