1
use std::{
2
    cmp::Ordering,
3
    collections::HashMap,
4
    error::Error as StdError,
5
    sync::{Arc, Mutex},
6
};
7

            
8
use async_trait::async_trait;
9
use hex;
10
use log::{error, warn};
11
use serde::{Deserialize, Serialize};
12
use serde_json::{self, Map, Value};
13
use tokio::task;
14
use url::Url;
15

            
16
use general_mq::{
17
    queue::{
18
        EventHandler as QueueEventHandler, GmqQueue, Message, MessageHandler, Status as QueueStatus,
19
    },
20
    Queue,
21
};
22
use sylvia_iot_corelib::{err, strings};
23

            
24
use super::{
25
    get_connection, new_data_queues, remove_connection, Connection, MgrMqStatus, MgrStatus, Options,
26
};
27

            
28
/// Uplink data from broker to application.
29
#[derive(Serialize)]
30
pub struct UlData {
31
    #[serde(rename = "dataId")]
32
    pub data_id: String,
33
    pub time: String,
34
    #[serde(rename = "pub")]
35
    pub publish: String,
36
    #[serde(rename = "deviceId")]
37
    pub device_id: String,
38
    #[serde(rename = "networkId")]
39
    pub network_id: String,
40
    #[serde(rename = "networkCode")]
41
    pub network_code: String,
42
    #[serde(rename = "networkAddr")]
43
    pub network_addr: String,
44
    #[serde(rename = "isPublic")]
45
    pub is_public: bool,
46
    pub profile: String,
47
    pub data: String,
48
    #[serde(skip_serializing_if = "Option::is_none")]
49
    pub extension: Option<Map<String, Value>>,
50
}
51

            
52
/// Downlink data from application to broker.
53
#[derive(Deserialize)]
54
pub struct DlData {
55
    #[serde(rename = "correlationId")]
56
    pub correlation_id: String,
57
    #[serde(rename = "deviceId")]
58
    pub device_id: Option<String>,
59
    #[serde(rename = "networkCode")]
60
    pub network_code: Option<String>,
61
    #[serde(rename = "networkAddr")]
62
    pub network_addr: Option<String>,
63
    pub data: String,
64
    pub extension: Option<Map<String, Value>>,
65
}
66

            
67
/// Downlink data response for [`DlData`].
68
#[derive(Default, Serialize)]
69
pub struct DlDataResp {
70
    #[serde(rename = "correlationId")]
71
    pub correlation_id: String,
72
    #[serde(rename = "dataId", skip_serializing_if = "Option::is_none")]
73
    pub data_id: Option<String>,
74
    #[serde(skip_serializing_if = "Option::is_none")]
75
    pub error: Option<String>,
76
    #[serde(skip_serializing_if = "Option::is_none")]
77
    pub message: Option<String>,
78
}
79

            
80
/// Downlink data result when processing or completing data transfer to the device.
81
#[derive(Serialize)]
82
pub struct DlDataResult {
83
    #[serde(rename = "dataId")]
84
    pub data_id: String,
85
    pub status: i32,
86
    #[serde(skip_serializing_if = "Option::is_none")]
87
    pub message: Option<String>,
88
}
89

            
90
/// The manager for application queues.
91
#[derive(Clone)]
92
pub struct ApplicationMgr {
93
    opts: Arc<Options>,
94

            
95
    // Information for delete connection automatically.
96
    conn_pool: Arc<Mutex<HashMap<String, Connection>>>,
97
    host_uri: String,
98

            
99
    uldata: Arc<Mutex<Queue>>,
100
    dldata: Arc<Mutex<Queue>>,
101
    dldata_resp: Arc<Mutex<Queue>>,
102
    dldata_result: Arc<Mutex<Queue>>,
103

            
104
    status: Arc<Mutex<MgrStatus>>,
105
    handler: Arc<Mutex<Arc<dyn EventHandler>>>,
106
}
107

            
108
/// Event handler trait for the [`ApplicationMgr`].
109
#[async_trait]
110
pub trait EventHandler: Send + Sync {
111
    async fn on_status_change(&self, mgr: &ApplicationMgr, status: MgrStatus);
112

            
113
    async fn on_dldata(
114
        &self,
115
        mgr: &ApplicationMgr,
116
        data: Box<DlData>,
117
    ) -> Result<Box<DlDataResp>, ()>;
118
}
119

            
120
/// The event handler for [`general_mq::queue::GmqQueue`].
121
struct MgrMqEventHandler {
122
    mgr: ApplicationMgr,
123
}
124

            
125
const QUEUE_PREFIX: &'static str = "broker.application";
126

            
127
impl ApplicationMgr {
128
    /// To create a manager instance.
129
54
    pub fn new(
130
54
        conn_pool: Arc<Mutex<HashMap<String, Connection>>>,
131
54
        host_uri: &Url,
132
54
        opts: Options,
133
54
        handler: Arc<dyn EventHandler>,
134
54
    ) -> Result<Self, String> {
135
54
        if opts.unit_id.len() == 0 {
136
7
            return Err("`unit_id` cannot be empty for application".to_string());
137
47
        }
138

            
139
47
        let conn = get_connection(&conn_pool, host_uri)?;
140

            
141
41
        let (uldata, dldata, dldata_resp, dldata_result) =
142
47
            new_data_queues(&conn, &opts, QUEUE_PREFIX, false)?;
143

            
144
41
        let mgr = ApplicationMgr {
145
41
            opts: Arc::new(opts),
146
41
            conn_pool,
147
41
            host_uri: host_uri.to_string(),
148
41
            uldata,
149
41
            dldata,
150
41
            dldata_resp: dldata_resp.unwrap(),
151
41
            dldata_result,
152
41
            status: Arc::new(Mutex::new(MgrStatus::NotReady)),
153
41
            handler: Arc::new(Mutex::new(handler)),
154
41
        };
155
41
        let mq_handler = Arc::new(MgrMqEventHandler { mgr: mgr.clone() });
156
41
        let mut q = { mgr.uldata.lock().unwrap().clone() };
157
41
        q.set_handler(mq_handler.clone());
158
41
        if let Err(e) = q.connect() {
159
            return Err(e.to_string());
160
41
        }
161
41
        let mut q = { mgr.dldata.lock().unwrap().clone() };
162
41
        q.set_handler(mq_handler.clone());
163
41
        q.set_msg_handler(mq_handler.clone());
164
41
        if let Err(e) = q.connect() {
165
            return Err(e.to_string());
166
41
        }
167
41
        let mut q = { mgr.dldata_resp.lock().unwrap().clone() };
168
41
        q.set_handler(mq_handler.clone());
169
41
        if let Err(e) = q.connect() {
170
            return Err(e.to_string());
171
41
        }
172
41
        let mut q = { mgr.dldata_result.lock().unwrap().clone() };
173
41
        q.set_handler(mq_handler.clone());
174
41
        if let Err(e) = q.connect() {
175
            return Err(e.to_string());
176
41
        }
177
41
        match conn {
178
29
            Connection::Amqp(_, counter) => {
179
29
                *counter.lock().unwrap() += 4;
180
29
            }
181
12
            Connection::Mqtt(_, counter) => {
182
12
                *counter.lock().unwrap() += 4;
183
12
            }
184
        }
185
41
        Ok(mgr)
186
54
    }
187

            
188
    /// The associated unit ID of the application.
189
68
    pub fn unit_id(&self) -> &str {
190
68
        self.opts.unit_id.as_str()
191
68
    }
192

            
193
    /// The associated unit code of the application.
194
27
    pub fn unit_code(&self) -> &str {
195
27
        self.opts.unit_code.as_str()
196
27
    }
197

            
198
    /// The application ID.
199
14
    pub fn id(&self) -> &str {
200
14
        self.opts.id.as_str()
201
14
    }
202

            
203
    /// The application code.
204
14
    pub fn name(&self) -> &str {
205
14
        self.opts.name.as_str()
206
14
    }
207

            
208
    /// Manager status.
209
115
    pub fn status(&self) -> MgrStatus {
210
115
        *self.status.lock().unwrap()
211
115
    }
212

            
213
    /// Detail status of each message queue.
214
4
    pub fn mq_status(&self) -> MgrMqStatus {
215
4
        MgrMqStatus {
216
4
            uldata: { self.uldata.lock().unwrap().status() },
217
4
            dldata: { self.dldata.lock().unwrap().status() },
218
4
            dldata_resp: { self.dldata_resp.lock().unwrap().status() },
219
4
            dldata_result: { self.dldata_result.lock().unwrap().status() },
220
4
            ctrl: QueueStatus::Closed,
221
4
        }
222
4
    }
223

            
224
    /// To close the manager queues.
225
49
    pub async fn close(&self) -> Result<(), Box<dyn StdError + Send + Sync>> {
226
49
        let mut q = { self.uldata.lock().unwrap().clone() };
227
49
        q.close().await?;
228
49
        let mut q = { self.dldata.lock().unwrap().clone() };
229
49
        q.close().await?;
230
49
        let mut q = { self.dldata_resp.lock().unwrap().clone() };
231
49
        q.close().await?;
232
49
        let mut q = { self.dldata_result.lock().unwrap().clone() };
233
49
        q.close().await?;
234

            
235
49
        remove_connection(&self.conn_pool, &self.host_uri, 4).await
236
49
    }
237

            
238
    /// Send uplink data to the application.
239
42
    pub fn send_uldata(&self, data: &UlData) -> Result<(), Box<dyn StdError>> {
240
42
        let payload = serde_json::to_vec(data)?;
241
42
        let queue = { (*self.uldata.lock().unwrap()).clone() };
242
42
        task::spawn(async move {
243
42
            let _ = queue.send_msg(payload).await;
244
42
        });
245
42
        Ok(())
246
42
    }
247

            
248
    /// Send downlink response for a downlink data to the application.
249
34
    pub async fn send_dldata_resp(
250
34
        &self,
251
34
        data: &DlDataResp,
252
34
    ) -> Result<(), Box<dyn StdError + Send + Sync>> {
253
34
        let payload = serde_json::to_vec(data)?;
254
34
        let queue = { (*self.dldata_resp.lock().unwrap()).clone() };
255
34
        queue.send_msg(payload).await
256
34
    }
257

            
258
    /// Send the downlink data process result to the application.
259
22
    pub async fn send_dldata_result(
260
22
        &self,
261
22
        data: &DlDataResult,
262
22
    ) -> Result<(), Box<dyn StdError + Send + Sync>> {
263
22
        let payload = serde_json::to_vec(data)?;
264
22
        let queue = { (*self.dldata_result.lock().unwrap()).clone() };
265
22
        queue.send_msg(payload).await
266
22
    }
267
}
268

            
269
#[async_trait]
270
impl QueueEventHandler for MgrMqEventHandler {
271
    async fn on_error(&self, _queue: Arc<dyn GmqQueue>, _err: Box<dyn StdError + Send + Sync>) {}
272

            
273
250
    async fn on_status(&self, _queue: Arc<dyn GmqQueue>, _status: QueueStatus) {
274
250
        let status = match { self.mgr.uldata.lock().unwrap().status() } == QueueStatus::Connected
275
76
            && { self.mgr.dldata.lock().unwrap().status() } == QueueStatus::Connected
276
38
            && { self.mgr.dldata_resp.lock().unwrap().status() } == QueueStatus::Connected
277
32
            && { self.mgr.dldata_result.lock().unwrap().status() } == QueueStatus::Connected
278
        {
279
227
            false => MgrStatus::NotReady,
280
23
            true => MgrStatus::Ready,
281
        };
282

            
283
250
        let mut changed = false;
284
250
        {
285
250
            let mut mutex = self.mgr.status.lock().unwrap();
286
250
            if *mutex != status {
287
42
                *mutex = status;
288
42
                changed = true;
289
208
            }
290
        }
291
250
        if changed {
292
42
            let handler = { self.mgr.handler.lock().unwrap().clone() };
293
42
            handler.on_status_change(&self.mgr, status).await;
294
208
        }
295
500
    }
296
}
297

            
298
#[async_trait]
299
impl MessageHandler for MgrMqEventHandler {
300
    // Validate and decode data.
301
36
    async fn on_message(&self, queue: Arc<dyn GmqQueue>, msg: Box<dyn Message>) {
302
        const FN_NAME: &'static str = "ApplicationMgr.on_message";
303

            
304
36
        let queue_name = queue.name();
305
36
        if queue_name.cmp(self.mgr.dldata.lock().unwrap().name()) == Ordering::Equal {
306
36
            let data = match parse_dldata_msg(msg.payload()) {
307
16
                Err(resp) => {
308
16
                    warn!("[{}] invalid format from {}", FN_NAME, queue_name);
309
16
                    if let Err(e) = msg.ack().await {
310
                        error!("[{}] ACK message error: {}", FN_NAME, e);
311
16
                    }
312
16
                    if let Err(e) = self.mgr.send_dldata_resp(&resp).await {
313
                        error!("[{}] send response error: {}", FN_NAME, e);
314
16
                    }
315
16
                    return;
316
                }
317
20
                Ok(data) => data,
318
20
            };
319
20
            let handler = { self.mgr.handler.lock().unwrap().clone() };
320
20
            match handler.on_dldata(&self.mgr, Box::new(data)).await {
321
                Err(_) => {
322
2
                    if let Err(e) = msg.nack().await {
323
                        error!("[{}] NACK message error: {}", FN_NAME, e);
324
2
                    }
325
                }
326
18
                Ok(resp) => {
327
18
                    if let Err(e) = msg.ack().await {
328
                        error!("[{}] ACK message error: {}", FN_NAME, e);
329
18
                    }
330
18
                    if let Err(e) = self.mgr.send_dldata_resp(resp.as_ref()).await {
331
                        error!("[{}] send response error: {}", FN_NAME, e);
332
18
                    }
333
                }
334
            }
335
        }
336
72
    }
337
}
338

            
339
/// Parses downlink data from the application and responds error for wrong format data.
340
36
fn parse_dldata_msg(msg: &[u8]) -> Result<DlData, DlDataResp> {
341
36
    let mut data = match serde_json::from_slice::<DlData>(msg) {
342
        Err(_) => {
343
2
            return Err(DlDataResp {
344
2
                correlation_id: "".to_string(),
345
2
                error: Some(err::E_PARAM.to_string()),
346
2
                message: Some("invalid format".to_string()),
347
2
                ..Default::default()
348
2
            });
349
        }
350
34
        Ok(data) => data,
351
34
    };
352
34

            
353
34
    if data.correlation_id.len() == 0 {
354
2
        return Err(DlDataResp {
355
2
            correlation_id: data.correlation_id.clone(),
356
2
            error: Some(err::E_PARAM.to_string()),
357
2
            message: Some("invalid `correlationId`".to_string()),
358
2
            ..Default::default()
359
2
        });
360
32
    }
361
32
    match data.device_id.as_ref() {
362
        None => {
363
21
            match data.network_code.as_ref() {
364
                None => {
365
2
                    return Err(DlDataResp {
366
2
                        correlation_id: data.correlation_id.clone(),
367
2
                        error: Some(err::E_PARAM.to_string()),
368
2
                        message: Some("missing `networkCode`".to_string()),
369
2
                        ..Default::default()
370
2
                    });
371
                }
372
19
                Some(code) => {
373
19
                    let code = code.to_lowercase();
374
19
                    match strings::is_code(code.as_str()) {
375
                        false => {
376
2
                            return Err(DlDataResp {
377
2
                                correlation_id: data.correlation_id.clone(),
378
2
                                error: Some(err::E_PARAM.to_string()),
379
2
                                message: Some("invalid `networkCode`".to_string()),
380
2
                                ..Default::default()
381
2
                            });
382
                        }
383
                        true => {
384
17
                            data.network_code = Some(code);
385
17
                            ()
386
17
                        }
387
17
                    }
388
17
                }
389
17
            }
390
17
            match data.network_addr.as_ref() {
391
                None => {
392
2
                    return Err(DlDataResp {
393
2
                        correlation_id: data.correlation_id.clone(),
394
2
                        error: Some(err::E_PARAM.to_string()),
395
2
                        message: Some("missing `networkAddr`".to_string()),
396
2
                        ..Default::default()
397
2
                    });
398
                }
399
15
                Some(addr) => match addr.len() {
400
                    0 => {
401
2
                        return Err(DlDataResp {
402
2
                            correlation_id: data.correlation_id.clone(),
403
2
                            error: Some(err::E_PARAM.to_string()),
404
2
                            message: Some("invalid `networkAddr`".to_string()),
405
2
                            ..Default::default()
406
2
                        });
407
                    }
408
                    _ => {
409
13
                        data.network_addr = Some(addr.to_lowercase());
410
13
                        ()
411
                    }
412
                },
413
            }
414
        }
415
11
        Some(device_id) => match device_id.len() {
416
            0 => {
417
2
                return Err(DlDataResp {
418
2
                    correlation_id: data.correlation_id.clone(),
419
2
                    error: Some(err::E_PARAM.to_string()),
420
2
                    message: Some("invalid `deviceId`".to_string()),
421
2
                    ..Default::default()
422
2
                });
423
            }
424
9
            _ => (),
425
        },
426
    }
427
22
    if data.data.len() > 0 {
428
22
        if let Err(_) = hex::decode(data.data.as_str()) {
429
2
            return Err(DlDataResp {
430
2
                correlation_id: data.correlation_id.clone(),
431
2
                error: Some(err::E_PARAM.to_string()),
432
2
                message: Some("invalid `data`".to_string()),
433
2
                ..Default::default()
434
2
            });
435
20
        }
436
20
        data.data = data.data.to_lowercase();
437
    }
438
20
    Ok(data)
439
36
}