Creating Factory Actor with Asyncronsous WorkerBuilder

[jgould22]

A little preamble to avoid the XY problem.

I am trying to create a factory actor.

The intention of the factory actor is to create a group of actors that make requests to and handle responses from an external service. Each worker would represent a connection to the same external service.

The idea is I send the request message one of the Factory workers picks it up and makes the request.

I have created a sort of sample actor below.

// Pseudo code, might not compile

pub(crate) struct MyActor;

#[derive(Clone)]
pub struct MyActorState {
    client: MyClient,
}

pub struct MyActorArguments {
    pub id: String,
    pub host: String,
    pub port: u16,
    pub timeout: Option<Duration>,
}

pub enum MyActorMessage {
    A,
    B,
    C,
}
#[cfg(feature = "cluster")]
impl ractor::Message for MyActorMessage {}

#[async_trait::async_trait]
impl Actor for MyActor {
    type Msg = MyActorMessage;
    type State = MyActorState;
    type Arguments = MyActorArguments;

    async fn pre_start(
        &self,
        _myself: ActorRef<Self::Msg>,
        args: MyActorArguments,
    ) -> Result<Self::State, ActorProcessingErr> {

        // Init a client
        let client = MyClient(
            args.id,
            args.host,
            args.port,
            args.timeout
                .unwrap_or_else(|| Duration::from_secs(DEFAULT_CONNECT_TIMEOUT)),
        )
        .await?;

        Ok(MyActorState { client })
    }

    async fn handle(
        &self,
        _myself: ActorRef<Self::Msg>,
        message: Self::Msg,
        state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        match message {
            MyActorMessage::A => {
                state.client.request("A")
            }
            MyActorMessage::B => {
                state.client.request("B")
            }
            MyActorMessage::C => {
                state.client.request("C")
            }
        }
        Ok(())
    }

}

#[cfg(test)]
mod tests {
    use ractor::Actor;

    #[tokio::test]
    async fn create_test_connection_actor() {
        let (client, actor_handle) = Actor::spawn(
            Some("Test Actor".to_string()),
            MyActor,
            MyActorArguments {
                id: String::from("remote-host"),
                host: String::from("remote-host.com"),
                port: 8080,
                timeout: None,
            },
        )
        .await
        .expect("Failed to start connection actor");

        let res =ractor::cast!(client, MyActorMessage::A).unwrap();
    }
}

In order to create my factory actor I am was using the https://github.com/slawlor/ractor/blob/main/ractor/src/factory/tests/mod.rs as a template.

The problem I am running into is for the regular actor above I can pass the MyActorArguments and then create my client in the pre_start function however for the factory actor it appears I need to implement the WorkerBuilder trait on my actor.

This build function of this trait returns a worker actor. The problem is I cant seem to figure out how to pass MyActorArguments or init the client in the build function since my client is async and the build function is sync.

Where can I put my

        // Init a client
        let client = MyClient(
            args.id,
            args.host,
            args.port,
            args.timeout
                .unwrap_or_else(|| Duration::from_secs(DEFAULT_CONNECT_TIMEOUT)),
        )
        .await?;

so that each factory worker has a client it can use?

Any help or recommendation is greatly appreciated!

[slawlor]

You shouldn't be "init'ing" the worker at all, you just need to create an instance of the struct. The factory will take care of spawning + setting up your actor's pre_start routine. In your last example, move any "init" logic to pre_start which is async, and just have the MyClient struct be just the arguments necessary to startup the actor.

[jgould22]

Thanks for the reply, I appreciate the help.

I think I made a mistake in my code sample the MyClient in this case is not an actor, but a client from another crate.

        let client = MyClient::new(
            args.id,
            args.host,
            args.port,
            args.timeout
                .unwrap_or_else(|| Duration::from_secs(DEFAULT_CONNECT_TIMEOUT)),
        )
        .await?;

If I take https://github.com/slawlor/ractor/blob/main/ractor/src/factory/tests/mod.rs#L75. In order to store MyClient in my basic actor I have been putting it in the actor state and initializing it in pre_start using the arguments passed to the spawn, which seems to work. Then in my handle function I can access it from the state of the actor. The state in the handle function of TestWorker on the other hand holds the WorkerStartContext which it appears to use to access a reference to the factory so it can return pings and ack jobs etc.

So I am trying to get my client accessible in the same way the counter is in the handle function, however unlike the counter in the TestWorker my clients initialization is both async and fallible and I was hoping that I could have the actors initialize them at start time instead of initializing a bunch and passing them to the builder as the counters are in the test.

Is this approach correct or am I misunderstanding ?

I will see if I can get a more complete example together in case my explanation is not clear.

[slawlor]

I think I understand what you're going for, but it's kind of against the notion of a factory. You need something, which is async initialized, available to each worker. The problem is that the WorkerBuilder is invoked from the factory's context, so if that operation were to fail to properly initialize a worker, the whole factory would crash, rather than a single worker (which could be restarted).

Think timeouts, spurious network issues, etc. All that shouldn't take down the whole factory. Instead each worker should be able to

1. Startup whatever async client it needs itself from some constant data propogated by the workerbuilder [or]
2. If there's like a global, single "client" then you can initialize it and have the WorkerBuilder hold a handle to it (since it's just a struct) [or]
3. Have the clients do an RPC to some external process which holds/manages these clients and can pass an initialized value to the worker.

The whole customization of the worker's instantiation is specifically guarded to avoid introducing crashes at the factory level. Does that all make sense?

[jgould22]

Ok I think understand, thank you for the explanation.

Option 1 would be similar to how the TestWorker is initializing its counters and passing them to the WorkerBuilder

Option 2 would be initializing the client in an Arc and then have each worker share the client amongst themselves where the WorkerBuilder just clones the Arc when it builds a new actor.

Option 3 would be like having a "ClientBuilder" actor or process that can produce initialized clients outside of the factories context, so that if it fails for whatever reason the factory does not also fail.

I will play around with the options and see which might work best for me.