prosa 0.4.1

ProSA core
Documentation 
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
137

use std::sync::Arc;

use prosa_macros::proc_settings;
use serde::{Deserialize, Serialize};
use tracing::debug;

use crate::core::adaptor::{Adaptor, MaybeAsync};
use crate::core::error::{BusError, ProcError};
use crate::core::msg::{InternalMsg, Msg};
use crate::core::proc::{Proc, ProcBusParam, proc};

use super::adaptor::StubAdaptor;

extern crate self as prosa;

/// Stub settings to list all services tu stub
#[proc_settings]
#[derive(Default, Debug, Deserialize, Serialize, Clone)]
pub struct StubSettings {
    /// Services to respond to
    pub service_names: Vec<String>,
}

impl StubSettings {
    /// Create a new Stub settings
    pub fn new(service_names: Vec<String>) -> StubSettings {
        StubSettings {
            service_names,
            ..Default::default()
        }
    }

    /// Method to add service name
    pub fn add_service_name(&mut self, service_name: String) {
        self.service_names.push(service_name);
    }
}

/// Stub processor to respond to a request
///
/// ```
/// use prosa::core::main::{MainProc, MainRunnable};
/// use prosa::core::proc::{proc, Proc, ProcBusParam, ProcConfig};
/// use prosa::stub::adaptor::StubParotAdaptor;
/// use prosa::stub::proc::{StubProc, StubSettings};
/// use prosa_utils::config::observability::Observability;
/// use prosa_utils::msg::simple_string_tvf::SimpleStringTvf;
/// use prosa::core::settings::settings;
/// use serde::Serialize;
///
/// // Main settings
/// #[settings]
/// #[derive(Default, Debug, Serialize)]
/// struct Settings {}
///
/// // Create bus and main processor
/// let settings = Settings::default();
/// let (bus, main) = MainProc::<SimpleStringTvf>::create(&settings, Some(1));
///
/// // Launch a stub processor
/// let stub_settings = StubSettings::new(vec![String::from("STUB_TEST")]);
/// let stub_proc = StubProc::<SimpleStringTvf>::create(1, "STUB_PROC".to_string(), bus.clone(), stub_settings);
/// Proc::<StubParotAdaptor>::run(stub_proc);
///
/// // Wait on main task
/// //main_task.await;
/// ```
#[proc(settings = prosa::stub::proc::StubSettings)]
pub struct StubProc {}

#[proc]
impl<A> Proc<A> for StubProc
where
    A: 'static + Adaptor + StubAdaptor<M> + std::marker::Send + std::marker::Sync,
{
    async fn internal_run(&mut self) -> Result<(), Box<dyn ProcError + Send + Sync>> {
        // Initiate an adaptor for the stub processor
        let adaptor = Arc::new(A::new(self)?);

        // Declare the processor
        self.proc.add_proc().await?;

        // Add all service to listen
        self.proc
            .add_service_proc(self.settings.service_names.clone())
            .await?;

        loop {
            if let Some(msg) = self.internal_rx_queue.recv().await {
                match msg {
                    InternalMsg::Request(mut msg) => {
                        let request_data = msg.take_data().ok_or(BusError::NoData)?;
                        let enter_span = msg.enter_span();

                        debug!(name: "stub_proc_request", target: "prosa::stub::proc", parent: msg.get_span(), proc_name = self.name(), stub_service = msg.get_service(), "{:?}", msg.get_data());

                        match adaptor.process_request(msg.get_service(), request_data) {
                            MaybeAsync::Ready(resp) => {
                                debug!(name: "stub_proc_response", target: "prosa::stub::proc", parent: msg.get_span(), stub_service = msg.get_service(), "{resp:?}");
                                drop(enter_span);
                                let _ = msg.return_result_to_sender(resp);
                            }
                            MaybeAsync::Future(future_resp) => {
                                drop(enter_span);
                                tokio::spawn(async move {
                                    let enter_span = msg.enter_span();
                                    let resp = future_resp.await;
                                    debug!(name: "stub_proc_response", target: "prosa::stub::proc", parent: msg.get_span(), stub_service = msg.get_service(), "{resp:?}");
                                    drop(enter_span);
                                    let _ = msg.return_result_to_sender(resp);
                                });
                            }
                        }
                    }
                    InternalMsg::Response(msg) => panic!(
                        "The stub processor {} receive a response {:?}",
                        self.get_proc_id(),
                        msg
                    ),
                    InternalMsg::Error(err) => panic!(
                        "The stub processor {} receive an error {:?}",
                        self.get_proc_id(),
                        err
                    ),
                    InternalMsg::Command(_) => todo!(),
                    InternalMsg::Config => todo!(),
                    InternalMsg::Service(table) => self.service = table,
                    InternalMsg::Shutdown => {
                        adaptor.terminate();
                        self.proc.remove_proc(None).await?;
                        return Ok(());
                    }
                }
            }
        }
    }
}