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
138
139
140
141
142
143

use std::{collections::HashMap, error::Error as StdError, sync::Arc};

use actix_web::{dev::HttpServiceFactory, error, web};
use general_mq::Queue;
use reqwest;

use sylvia_iot_corelib::{constants::DbEngine, err::ErrResp};

use crate::{
    libs::{
        config::{self, Config},
        mq::{self, Connection},
    },
    models::{self, ConnOptions, Model, MongoDbOptions, SqliteOptions},
};

pub mod middleware;
mod v1;

/// The resources used by this service.
#[derive(Clone)]
pub struct State {
    /// The scope root path for the service.
    ///
    /// For example `data`, the APIs are
    /// - `http://host:port/data/api/v1/application-uldata/xxx`
    /// - `http://host:port/data/api/v1/network-uldata/xxx`
    pub scope_path: &'static str,
    /// The database model.
    pub model: Arc<dyn Model>,
    /// The sylvia-iot-auth base API path with host.
    ///
    /// For example, `http://localhost:1080/auth`.
    pub auth_base: String,
    /// The sylvia-iot-broker base API path with host.
    ///
    /// For example, `http://localhost:2080/broker`.
    pub broker_base: String,
    /// The client for internal HTTP requests.
    pub client: reqwest::Client,
    /// Queue connections. Key is uri.
    pub mq_conns: HashMap<String, Connection>,
    /// Data channel receivers. Key is data channel name such as `broker.data`, `coremgr.data`, ...
    pub data_receivers: HashMap<String, Queue>,
}

/// The sylvia-iot module specific error codes in addition to standard [`ErrResp`].
pub struct ErrReq;

impl ErrReq {
    pub const UNIT_NOT_EXIST: (u16, &'static str) = (400, "err_data_unit_not_exist");
    pub const USER_NOT_EXIST: (u16, &'static str) = (400, "err_data_user_not_exist");
}

/// To create resources for the service.
pub async fn new_state(
    scope_path: &'static str,
    conf: &Config,
) -> Result<State, Box<dyn StdError>> {
    let conf = config::apply_default(conf);
    let db_opts = match conf.db.as_ref().unwrap().engine.as_ref().unwrap().as_str() {
        DbEngine::MONGODB => {
            let conf = conf.db.as_ref().unwrap().mongodb.as_ref().unwrap();
            ConnOptions::MongoDB(MongoDbOptions {
                url: conf.url.as_ref().unwrap().to_string(),
                db: conf.database.as_ref().unwrap().to_string(),
                pool_size: conf.pool_size,
            })
        }
        _ => {
            let conf = conf.db.as_ref().unwrap().sqlite.as_ref().unwrap();
            ConnOptions::Sqlite(SqliteOptions {
                path: conf.path.as_ref().unwrap().to_string(),
            })
        }
    };
    let model = models::new(&db_opts).await?;
    let auth_base = conf.auth.as_ref().unwrap().clone();
    let broker_base = conf.broker.as_ref().unwrap().clone();
    let mut mq_conns = HashMap::new();
    let ch_conf = conf.mq_channels.as_ref().unwrap();
    let data_receivers = new_data_receivers(&model, &mut mq_conns, ch_conf)?;
    let state = State {
        scope_path,
        model,
        auth_base,
        broker_base,
        client: reqwest::Client::new(),
        mq_conns,
        data_receivers,
    };
    Ok(state)
}

/// To register service URIs in the specified root path.
pub fn new_service(state: &State) -> impl HttpServiceFactory {
    web::scope(state.scope_path)
        .app_data(web::JsonConfig::default().error_handler(|err, _| {
            error::ErrorBadRequest(ErrResp::ErrParam(Some(err.to_string())))
        }))
        .app_data(web::QueryConfig::default().error_handler(|err, _| {
            error::ErrorBadRequest(ErrResp::ErrParam(Some(err.to_string())))
        }))
        .app_data(web::Data::new(state.clone()))
        .service(v1::application_uldata::new_service(
            "/api/v1/application-uldata",
            state,
        ))
        .service(v1::application_dldata::new_service(
            "/api/v1/application-dldata",
            state,
        ))
        .service(v1::network_uldata::new_service(
            "/api/v1/network-uldata",
            state,
        ))
        .service(v1::network_dldata::new_service(
            "/api/v1/network-dldata",
            state,
        ))
        .service(v1::coremgr_opdata::new_service(
            "/api/v1/coremgr-opdata",
            state,
        ))
}

pub fn new_data_receivers(
    model: &Arc<dyn Model>,
    mq_conns: &mut HashMap<String, Connection>,
    ch_conf: &config::MqChannels,
) -> Result<HashMap<String, Queue>, Box<dyn StdError>> {
    let mut data_receivers = HashMap::<String, Queue>::new();

    let conf = ch_conf.broker.as_ref().unwrap();
    let q = mq::broker::new(model.clone(), mq_conns, &conf)?;
    data_receivers.insert("broker.data".to_string(), q);

    let conf = ch_conf.coremgr.as_ref().unwrap();
    let q = mq::coremgr::new(model.clone(), mq_conns, &conf)?;
    data_receivers.insert("coremgr.data".to_string(), q);

    Ok(data_receivers)
}