How to deserialize messages within Actix actors?

Question

My intention is receiving events through WebSockets and use them on the closures of main. This works when the messages are pure text (String), but the idea is deserializing that text into some structs.

In this example I've added only Data, Error and Event, but in other cases, it could be different, so I've used generics to do that, but I'm a little lost. The compiler has suggested several things that I've tried, but I don't know how to "force" that the message is casted into a specific type (Data in this example, but EventManager could be used on other parts, so it should be generic).

I've attached this code, that tries to show my idea, although it doesn't compile:

events.rs:

use actix::*;
use actix_web::ws::{Client, Message, ProtocolError};
use futures::Future;

use serde::de;
use serde_json::from_str;

struct MyActor<T> {
    manager: EventManager<T>,
}

impl<T: 'static> Actor for MyActor<T> {
    type Context = Context<Self>;
}

impl<T: 'static> StreamHandler<Message, ProtocolError> for MyActor<T> {
    fn handle(&mut self, msg: Message, _ctx: &mut Context<Self>) {
        match msg {
            Message::Text(text) => {
                debug!("Received {}", text);

                for idx in 0..self.manager.events.len() {
                    let data =
                        from_str(&text).expect(&format!("Error when deserializing {:?}", text));
                    (self.manager.events[idx].handler)(data)
                }
            }
            _ => panic!(),
        }
    }
}

pub struct Event<T> {
    handler: Box<Fn(T) + 'static>,
}

pub struct EventManager<T> {
    events: Vec<Event<T>>,
}

impl<T: 'static> EventManager<T>
where
    T: serde::Deserialize<'static>,
{
    pub fn new() -> Self {
        Self { events: vec![] }
    }

    pub fn capture<F>(&mut self, function: F)
    where
        F: for<'h> Fn(T) + 'static,
    {
        let event = Event {
            handler: Box::new(function),
        };
        self.events.push(event);
    }

    pub fn run(self) {
        let runner = System::new("example");

        debug!("run");

        Arbiter::spawn(
            Client::new("example")
                .connect()
                .map(|(reader, _writer)| {
                    MyActor::create(|ctx| {
                        MyActor::add_stream(reader, ctx);
                        MyActor { manager: self }
                    });
                })
                .map_err(|err| {}),
        );

        runner.run();
    }
}

main.rs:

#[macro_use]
extern crate log;
extern crate actix;
extern crate actix_web;
extern crate env_logger;
extern crate futures;
extern crate serde;
#[macro_use]
extern crate serde_derive;
extern crate serde_json;

pub mod events;

use actix::*;
use serde::de;
use serde::de::{Deserialize, Deserializer};

use events::EventManager;

#[derive(Debug, Message, Serialize, Deserialize)]
#[serde(untagged)]
pub enum Data {
    Error(Error),
    Event(Event),
}

#[derive(Debug, Serialize, Deserialize)]
pub struct Error {
    message: String,
    code: String,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct Event {
    name: String,
    content: String,
}

fn main() {
    env_logger::init();

    let mut client = EventManager::<Data>new();

    client.capture(|data| debug!("event: {:?}", data));
    client.run();
}

All the code could be see in https://github.com/foochi/how-deserialize-within-actix

attdona · Accepted Answer · 2018-10-16T17:47:36.100

There are some fixes to get it compile.

The trick to get it compile is to use Higher-Rank Trait Bounds (HTRB) trait bounds instead of declaring 'static lifetimes.

Follow the compiler suggestion and bind the T: serde::Deserialize<'_> trait:

impl<T> StreamHandler<Message, ProtocolError> for MyActor<T>
where
    for<'de> T: serde::Deserialize<'de> + 'static,

Then change also the Deserialize<'static> trait bound associated to theEventManager impl with a HTRB trait bound to get it compatible with the requirement of StreamHandler implementation:

impl<T: 'static> EventManager<T>
where
    for<'de> T: serde::Deserialize<'de>,

Finally, if you correct the line the create the client with the right sintax:

let mut client: EventManager<Data> = EventManager::new();

the example code should compile.

Note: For capture the use of a Higher Trait Bound for declaring the Fn requirement is redundant, do simply:

pub fn capture<F>(&mut self, function: F)
where
    F: Fn(T) + 'static,

Thank you. This other question, could be useful for understanding HTRB: https://stackoverflow.com/questions/35592750/how-does-for-syntax-differ-from-a-regular-lifetime-bound/35595491#35595491 — Deveres, Oct 20 '18 at 16:21

How to deserialize messages within Actix actors?

1 Answers1