-
Notifications
You must be signed in to change notification settings - Fork 27
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
fbf6bc2
commit 300e0ca
Showing
7 changed files
with
242 additions
and
12 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| use crate::actor::context::ActorContext; | ||
| use crate::actor::message::{Handler, Message}; | ||
| use crate::actor::{Actor, ActorRefErr, IntoActor, LocalActorRef}; | ||
| use crate::remote::cluster::singleton::factory::SingletonFactory; | ||
| use crate::remote::cluster::singleton::manager::{Manager, State}; | ||
|
|
||
| pub struct ActorStarted<A: Actor> { | ||
| actor_ref: LocalActorRef<A>, | ||
| } | ||
|
|
||
| pub struct ActorFailure { | ||
| error: ActorRefErr, | ||
| } | ||
|
|
||
| impl<A: Actor> Message for ActorStarted<A> { | ||
| type Result = (); | ||
| } | ||
|
|
||
| impl Message for ActorFailure { | ||
| type Result = (); | ||
| } | ||
|
|
||
| impl<F: SingletonFactory> Manager<F> { | ||
| pub async fn start_actor(&mut self, ctx: &ActorContext) { | ||
| let state = self.factory.create(); | ||
| let sys = self.sys.actor_system().clone(); | ||
| let manager_ref = self.actor_ref(ctx); | ||
| let actor_id = self.singleton_actor_id.clone(); | ||
| let _ = tokio::spawn(async move { | ||
| let actor = state.into_actor(Some(actor_id), &sys).await; | ||
| match actor { | ||
| Ok(actor_ref) => { | ||
| let _ = manager_ref.notify(ActorStarted { actor_ref }); | ||
| } | ||
| Err(error) => { | ||
| let _ = manager_ref.notify(ActorFailure { error }); | ||
| } | ||
| } | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| #[async_trait] | ||
| impl<F: SingletonFactory> Handler<ActorStarted<F::Actor>> for Manager<F> { | ||
| async fn handle(&mut self, message: ActorStarted<F::Actor>, ctx: &mut ActorContext) { | ||
| match &self.state { | ||
| State::Starting { .. } => { | ||
| info!("singleton actor started"); | ||
| } | ||
| _ => { | ||
| warn!("Invalid state, expected `Starting`"); | ||
| } | ||
| } | ||
|
|
||
| let actor_ref = message.actor_ref; | ||
| self.state = State::Running { actor_ref } | ||
|
|
||
| // TODO: broadcast to all managers that we're running the actor | ||
| } | ||
| } | ||
|
|
||
| #[async_trait] | ||
| impl<F: SingletonFactory> Handler<ActorFailure> for Manager<F> { | ||
| async fn handle(&mut self, message: ActorFailure, ctx: &mut ActorContext) { | ||
| error!("Actor start failed (error={})", message.error); | ||
| // TODO: retry start?? | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| use std::time::Duration; | ||
| use tokio::time::sleep; | ||
| use tracing::Level; | ||
| use coerce::actor::Actor; | ||
| use coerce::actor::system::ActorSystem; | ||
| use coerce::remote::cluster::singleton::SingletonBuilder; | ||
| use coerce::remote::system::RemoteActorSystem; | ||
|
|
||
| mod util; | ||
|
|
||
| struct SingletonActor { | ||
|
|
||
| } | ||
|
|
||
| impl Actor for SingletonActor { | ||
|
|
||
| } | ||
|
|
||
| #[tokio::test] | ||
| pub async fn test_cluster_singleton_start() { | ||
| util::create_logger(Some(Level::DEBUG)); | ||
|
|
||
| let system = ActorSystem::new(); | ||
| let system2 = ActorSystem::new(); | ||
| let remote = RemoteActorSystem::builder() | ||
| .with_tag("remote-1") | ||
| .with_id(1) | ||
| .with_actor_system(system) | ||
| .build() | ||
| .await; | ||
|
|
||
| let remote2 = RemoteActorSystem::builder() | ||
| .with_tag("remote-2") | ||
| .with_id(2) | ||
| .with_actor_system(system2) | ||
| .build() | ||
| .await; | ||
|
|
||
| remote | ||
| .clone() | ||
| .cluster_worker() | ||
| .listen_addr("localhost:30101") | ||
| .start() | ||
| .await; | ||
|
|
||
| remote2 | ||
| .clone() | ||
| .cluster_worker() | ||
| .listen_addr("localhost:30102") | ||
| .with_seed_addr("localhost:30101") | ||
| .start() | ||
| .await; | ||
|
|
||
| let singleton = SingletonBuilder::new(remote) | ||
| .factory(|| SingletonActor {}) | ||
| .build().await; | ||
|
|
||
| let singleton2 = SingletonBuilder::new(remote2) | ||
| .factory(|| SingletonActor {}) | ||
| .build().await; | ||
|
|
||
| sleep(Duration::from_secs(5)).await; | ||
| } |