aves-kea 0.5.0

Specialized knowledge domain for the Aves async application runtime
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

use std::path::PathBuf;

use aves::bus::{BusRpcContext, EventBus};
use aves::config::{
    ConfigBroadcast, ConfigDirectory, ConfigRegistrationSudoService, GetConfigSectionInput, RegisterSectionInput,
    SectionRegistration, dispatch_get_config_section,
};
use aves::flow::FlowWorkspaceSourceService;
use aves::runtime::{
    ConstructionFacade, DomainCallbacks, DomainDescriptorFactory, DomainFuture, ServicePublisher,
    dispatch_lifecycle_request,
};
use aves::utils::errors::SharedInfraError;

use super::documents::create_workspace_source;
use super::handle::Handle as KnowledgeHandle;
use super::messages::{KnowledgeBroadcast, KnowledgeRequest, KnowledgeRequestPayload, KnowledgeResponse};
use super::types::{KnowledgeConfig, KnowledgeError};

/// Empty lifecycle marker. Runtime state remains owned by the domain handle.
pub struct KnowledgeDomain;

aves::domain! {
    marker KnowledgeDomain;

    request KnowledgeRequestPayload => KnowledgeResponse
    error = KnowledgeError
    transport = SharedInfraError
    target = "knowledge";

    outbound KnowledgeResponse => KnowledgeBroadcast {
        KnowledgeCreated { knowledge: _ } => RecordsChanged,
        KnowledgeRecordsChanged => RecordsChanged,
        KnowledgeUpdated { knowledge: _ } => RecordsChanged,
        KnowledgeDeleted { result: _ } => RecordsChanged,
        CanvasDashboardCreated { dashboard } => CanvasDashboardCreated { dashboard_id: dashboard.id },
        CanvasDashboardUpdated { dashboard } => CanvasDashboardUpdated { dashboard_id: dashboard.id },
        CanvasDashboardDeleted { id, result: _ } => CanvasDashboardDeleted { dashboard_id: id },
    }
}

pub(super) use __Knowledge_generated_routing::KnowledgeGeneratedHandler;

impl KnowledgeDomain {
    async fn construct_and_run<Root>(
        bus: &mut EventBus<Root>,
        root_context: BusRpcContext,
        workspace_source_publisher: ServicePublisher<dyn FlowWorkspaceSourceService>,
        config: KnowledgeConfig,
        config_dir: PathBuf,
    ) where
        Root: Clone + Send + 'static,
        KnowledgeBroadcast: Into<Root> + Clone + Send + 'static,
        ConfigBroadcast: TryFrom<Root>,
    {
        let (workspace_source, shared_root) = create_workspace_source(&config, &config_dir);
        assert!(workspace_source_publisher.publish(workspace_source).is_ok());
        let handle = KnowledgeHandle::new(bus, root_context, config_dir, config, shared_root).await;
        tokio::spawn(async move { handle.run().await });
    }

    async fn start(_: BusRpcContext) {}

    async fn stop(context: BusRpcContext) {
        let _ = dispatch_lifecycle_request::<KnowledgeDomain, _, _, _>(context, |reply_to| KnowledgeRequest {
            reply_to,
            payload: KnowledgeRequestPayload::StopRuntime,
        })
        .await;
    }
}

impl<Root> DomainDescriptorFactory<EventBus<Root>> for KnowledgeDomain
where
    Root: Clone + Send + 'static,
    KnowledgeBroadcast: Into<Root> + Clone + Send + 'static,
    ConfigBroadcast: TryFrom<Root>,
{
    fn callbacks() -> DomainCallbacks<EventBus<Root>> {
        fn new_domain<'facade, 'bus, Root>(
            facade: &'facade mut ConstructionFacade<'bus, EventBus<Root>>,
        ) -> DomainFuture<'facade>
        where
            Root: Clone + Send + 'static,
            KnowledgeBroadcast: Into<Root> + Clone + Send + 'static,
            ConfigBroadcast: TryFrom<Root>,
        {
            let root_context = facade.root_context().clone();
            let publisher = facade.publisher::<dyn FlowWorkspaceSourceService>();

            let directory = facade
                .input::<ConfigDirectory>()
                .expect("config directory")
                .0
                .clone();

            let registration = facade.reader::<dyn ConfigRegistrationSudoService>();
            let bus = facade.bus_mut();

            Box::pin(async move {
                registration
                    .get()
                    .expect("Config registration service must be published first")
                    .register_section(RegisterSectionInput {
                        registration: SectionRegistration::new::<KnowledgeConfig>("knowledge")
                            .expect("valid Knowledge section"),
                    })
                    .await
                    .expect("Knowledge section must be unique");

                let config = dispatch_get_config_section(
                    &root_context,
                    GetConfigSectionInput {
                        section: "knowledge".to_owned(),
                    },
                )
                .await
                .expect("Knowledge configuration must load");

                let config = serde_json::from_value(config.value).expect("Knowledge section validated by Config");
                KnowledgeDomain::construct_and_run(bus, root_context, publisher, config, directory).await;
            })
        }

        DomainCallbacks {
            new: new_domain::<Root>,
            start: |context| Box::pin(KnowledgeDomain::start(context)),
            stop: |context| Box::pin(KnowledgeDomain::stop(context)),
        }
    }
}