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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

use crate::event::handler::EventHandlerBuilder;
use crate::event::EventHandler;
use crate::Application;
use crate::Router;
use crate::SubscriberManager;
use actix::Actor;
use event_store::prelude::Storage;
use futures::Future;
use std::any::TypeId;
use std::collections::HashMap;
use std::marker::PhantomData;
use std::pin::Pin;
use tracing::trace;

use super::InternalApplication;

#[async_trait::async_trait]
pub trait StorageConfig<S>
where
    S: Storage,
{
    async fn build(&self) -> S;
}

/// Struct to configure and launch an `Application` instance
///
/// An application can only be launch though an ApplicationBuilder.
///
/// This builder will have the responsability to configure every part of the application.
///
/// Note: In most cases you will not have to use `ApplicationBuilder` directly.
/// `Application` will be the root of everything.
pub struct ApplicationBuilder<A: Application> {
    app: std::marker::PhantomData<A>,
    storage: Pin<Box<dyn Future<Output = A::Storage>>>,
    event_handlers: Vec<Pin<Box<dyn Future<Output = ()>>>>,
    eventmapper: HashMap<String, TypeId>,
}

impl<A: Application> std::default::Default for ApplicationBuilder<A> {
    fn default() -> Self {
        Self {
            event_handlers: Vec::new(),
            app: std::marker::PhantomData,
            storage: Box::pin(async { A::Storage::default() }),
            eventmapper: HashMap::new(),
            // event_resolver: None,
        }
    }
}

impl<A> ApplicationBuilder<A>
where
    A: Application,
{
    /// Adds an EventHandler to the application
    pub fn event_handler<E: EventHandler + 'static>(mut self, handler: E) -> Self {
        self.event_handlers
            .push(Box::pin(EventHandlerBuilder::new(handler).register::<A>()));
        self
    }

    /// Adds a StorageConfig use later to create the event_store Storage
    pub fn with_storage_config<CFG: StorageConfig<A::Storage> + 'static>(
        mut self,
        storage: CFG,
    ) -> Self {
        self.storage = Box::pin(async move { storage.build().await });

        self
    }

    /// Adds a preconfigured Backend as Storage.
    /// The storage isn't start, only when the application is launched.
    pub fn storage<F: Future<Output = Result<A::Storage, E>> + 'static, E: std::fmt::Debug>(
        mut self,
        storage: F,
    ) -> Self {
        self.storage =
            Box::pin(async move { storage.await.expect("Unable to connect the storage") });

        self
    }

    /// Launch the application
    ///
    /// Meaning that:
    /// - The storage future is resolved and registered
    /// - An EventStore instance is started based on this storage
    /// - A Router is started to handle commands
    /// - A SubscriberManager is started to dispatch incomming events
    #[tracing::instrument(name = "Chekov::Launch", skip(self), fields(app = %A::get_name()))]
    pub async fn launch(self) {
        trace!(
            "Launching a new Chekov instance with {}",
            std::any::type_name::<A::Storage>()
        );

        let storage: A::Storage = self.storage.await;
        let event_store: event_store::EventStore<A::Storage> = event_store::EventStore::builder()
            .storage(storage)
            .build()
            .await
            .unwrap();

        use futures::future::join_all;

        join_all(self.event_handlers).await;

        let event_store_addr = event_store.start();

        let router = Router::<A> {
            _app: std::marker::PhantomData,
            _before_dispatch: vec![],
        };

        let addr = router.start();
        let subscriber_manager_addr =
            SubscriberManager::<A>::new(event_store_addr.clone(), self.eventmapper).start();

        let event_store = crate::event_store::EventStore::<A> {
            addr: event_store_addr,
        }
        .start();

        ::actix::SystemRegistry::set(event_store);
        ::actix::SystemRegistry::set(addr);
        ::actix::SystemRegistry::set(subscriber_manager_addr);
        ::actix::SystemRegistry::set(
            InternalApplication::<A> {
                _phantom: PhantomData,
            }
            .start(),
        );
    }
}