chekov 0.1.1

CQRS/ES Framework
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

use crate::{
    command::Command, command::CommandMetadatas, message::Dispatch, Application,
    CommandExecutorError,
};
use actix::{prelude::WrapFuture, SystemService};
use tracing::trace;

#[doc(hidden)]
#[derive(Default)]
pub struct Router<A: Application> {
    pub(crate) _app: std::marker::PhantomData<A>,
    pub(crate) _before_dispatch: Vec<String>,
}

impl<A: Application> ::actix::Actor for Router<A> {
    type Context = ::actix::Context<Self>;
}

impl<A: Application> ::actix::registry::SystemService for Router<A> {}
impl<A: Application> ::actix::Supervised for Router<A> {}

impl<A: Application, T: Command> ::actix::Handler<Dispatch<T, A>> for Router<A>
where
    <T as Command>::ExecutorRegistry: actix::Handler<Dispatch<T, A>>,
{
    type Result = actix::ResponseActFuture<Self, Result<Vec<T::Event>, CommandExecutorError>>;

    #[tracing::instrument(name = "Router", skip(self, _ctx, msg), fields(correlation_id = %msg.metadatas.correlation_id))]
    fn handle(&mut self, msg: Dispatch<T, A>, _ctx: &mut Self::Context) -> Self::Result {
        let to =
            <T::ExecutorRegistry as SystemService>::from_registry().recipient::<Dispatch<T, A>>();
        trace!(
            to = ::std::any::type_name::<T::ExecutorRegistry>(),
            "Route command",
        );
        Box::pin(async move { to.send(msg).await? }.into_actor(self))
    }
}

impl<A: Application> Router<A> {
    #[tracing::instrument(name = "Dispatcher", skip(cmd, metadatas), fields(correlation_id = %metadatas.correlation_id))]
    pub async fn dispatch<C: Command>(
        cmd: C,
        metadatas: CommandMetadatas,
    ) -> Result<Vec<C::Event>, CommandExecutorError>
    where
        <C as Command>::ExecutorRegistry: actix::Handler<Dispatch<C, A>>,
    {
        trace!(
            executor = ::std::any::type_name::<C::Executor>(),
            "Sending {} to Router",
            ::std::any::type_name::<C>()
        );

        Self::from_registry()
            .send(Dispatch::<C, A> {
                metadatas,
                storage: std::marker::PhantomData,
                command: cmd,
            })
            .await?
    }
}