Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Initial work on singleton proxy, simple
PipeTo/pipe_to to run fut…
…ures and pipe the result back to actor refs as a message + further improvements
- Loading branch information
1 parent
945c8aa
commit b225db4
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
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,91 @@ | ||
| use crate::actor::context::ActorContext; | ||
| use crate::actor::message::{Handler, Message}; | ||
| use crate::actor::{Actor, ActorRef, ToActorId}; | ||
| use crate::remote::cluster::singleton::proxy::send::Buffered; | ||
| use std::collections::VecDeque; | ||
|
|
||
| pub mod send; | ||
|
|
||
| pub enum ProxyState<A> { | ||
| Buffered { | ||
| request_queue: VecDeque<Box<dyn Buffered<A>>>, | ||
| }, | ||
|
|
||
| Active { | ||
| actor_ref: ActorRef<A>, | ||
| }, | ||
| } | ||
|
|
||
| pub struct Proxy<A: Actor> { | ||
| state: ProxyState<A>, | ||
| } | ||
|
|
||
| impl<A: Actor> Proxy<A> { | ||
| pub fn new() -> Self { | ||
| Self { | ||
| state: ProxyState::Buffered { | ||
| request_queue: VecDeque::new(), | ||
| }, | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl<A: Actor> ProxyState<A> { | ||
| pub fn is_active(&self) -> bool { | ||
| matches!(&self, &ProxyState::Active { .. }) | ||
| } | ||
| } | ||
|
|
||
| #[async_trait] | ||
| impl<A: Actor> Actor for Proxy<A> {} | ||
|
|
||
| pub struct SingletonStarted<A: Actor> { | ||
| actor_ref: ActorRef<A>, | ||
| } | ||
|
|
||
| pub struct SingletonStopping {} | ||
|
|
||
| impl<A> Message for SingletonStarted<A> { | ||
| type Result = (); | ||
| } | ||
|
|
||
| impl Message for SingletonStopping { | ||
| type Result = (); | ||
| } | ||
|
|
||
| #[async_trait] | ||
| impl<A: Actor> Handler<SingletonStarted<A>> for Proxy<A> { | ||
| async fn handle(&mut self, message: SingletonStarted<A>, ctx: &mut ActorContext) { | ||
| let actor_ref = message.actor_ref; | ||
|
|
||
| match &mut self.state { | ||
| ProxyState::Buffered { request_queue } => { | ||
| debug!( | ||
| "emitting {} buffered messages to {}", | ||
| request_queue.len(), | ||
| &actor_ref | ||
| ); | ||
|
|
||
| while let Some(mut buffered) = request_queue.pop_front() { | ||
| buffered.send(actor_ref.clone()); | ||
| } | ||
| } | ||
| _ => {} | ||
| } | ||
|
|
||
| self.state = ProxyState::Active { actor_ref }; | ||
| } | ||
| } | ||
|
|
||
| #[async_trait] | ||
| impl<A: Actor> Handler<SingletonStopping> for Proxy<A> { | ||
| async fn handle(&mut self, _: SingletonStopping, ctx: &mut ActorContext) { | ||
| debug!("singleton actor stopped, buffering messages"); | ||
|
|
||
| if self.state.is_active() { | ||
| self.state = ProxyState::Buffered { | ||
| request_queue: VecDeque::new(), | ||
| } | ||
| } | ||
| } | ||
| } | ||
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 crate::actor::context::ActorContext; | ||
| use crate::actor::message::{Handler, Message}; | ||
| use crate::actor::{Actor, ActorRef, ActorRefErr}; | ||
| use crate::remote::cluster::singleton::factory::SingletonFactory; | ||
| use crate::remote::cluster::singleton::proxy::{Proxy, ProxyState}; | ||
| use tokio::sync::oneshot::Sender; | ||
|
|
||
| pub struct Send<M: Message> { | ||
| message: Option<M>, | ||
| result_channel: Option<Sender<M::Result>>, | ||
| } | ||
|
|
||
| impl<M: Message> Send<M> { | ||
| pub fn deliver<A>(&mut self, actor: ActorRef<A>) | ||
| where | ||
| A: Handler<M>, | ||
| { | ||
| let message = self.message.take().unwrap(); | ||
| let result_channel = self.result_channel.take().unwrap(); | ||
| tokio::spawn(async move { | ||
| let res = actor.send(message).await; | ||
| match res { | ||
| Ok(r) => result_channel.send(r), | ||
| Err(_) => {} | ||
| } | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| pub trait Buffered<A: Actor>: 'static + Sync + std::marker::Send { | ||
| fn send(&mut self, actor_ref: ActorRef<A>); | ||
| } | ||
|
|
||
| impl<A: Actor, M: Message> Buffered<A> for Send<M> | ||
| where | ||
| A: Handler<M>, | ||
| { | ||
| fn send(&mut self, actor_ref: ActorRef<A>) { | ||
| self.deliver(actor_ref) | ||
| } | ||
| } | ||
|
|
||
| impl<M: Message> Message for Send<M> { | ||
| type Result = (); | ||
| } | ||
|
|
||
| #[async_trait] | ||
| impl<A: Actor, M: Message> Handler<Send<M>> for Proxy<A> | ||
| where | ||
| A: Handler<M>, | ||
| { | ||
| async fn handle(&mut self, mut message: Send<M>, ctx: &mut ActorContext) { | ||
| match &mut self.state { | ||
| ProxyState::Buffered { request_queue } => { | ||
| request_queue.push_back(Box::new(message)); | ||
| } | ||
|
|
||
| ProxyState::Active { actor_ref } => { | ||
| message.deliver(actor_ref.clone()); | ||
| } | ||
| } | ||
| } | ||
| } |