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

            
9
use async_trait::async_trait;
10
use chrono::{DateTime, Utc};
11
use hex;
12
use serde::{Deserialize, Serialize};
13
use serde_json::{self, Map, Value};
14
use tokio::task;
15
use url::Url;
16

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

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

            
29
/// Uplink data from broker to application.
30
pub struct UlData {
31
    pub data_id: String,
32
    pub time: DateTime<Utc>,
33
    pub publish: DateTime<Utc>,
34
    pub device_id: String,
35
    pub network_id: String,
36
    pub network_code: String,
37
    pub network_addr: String,
38
    pub is_public: bool,
39
    pub data: Vec<u8>,
40
    pub extension: Option<Map<String, Value>>,
41
}
42

            
43
/// Downlink data from application to broker.
44
pub struct DlData {
45
    pub correlation_id: String,
46
    pub device_id: Option<String>,
47
    pub network_code: Option<String>,
48
    pub network_addr: Option<String>,
49
    pub data: Vec<u8>,
50
    pub extension: Option<Map<String, Value>>,
51
}
52

            
53
/// Downlink data response for [`DlData`].
54
#[derive(Deserialize)]
55
pub struct DlDataResp {
56
    #[serde(rename = "correlationId")]
57
    pub correlation_id: String,
58
    #[serde(rename = "dataId")]
59
    pub data_id: Option<String>,
60
    pub error: Option<String>,
61
    pub message: Option<String>,
62
}
63

            
64
/// Downlink data result when processing or completing data transfer to the device.
65
#[derive(Deserialize)]
66
pub struct DlDataResult {
67
    #[serde(rename = "dataId")]
68
    pub data_id: String,
69
    pub status: i32,
70
    pub message: Option<String>,
71
}
72

            
73
/// The manager for application queues.
74
#[derive(Clone)]
75
pub struct ApplicationMgr {
76
    opts: Arc<Options>,
77

            
78
    // Information for delete connection automatically.
79
    conn_pool: Arc<Mutex<HashMap<String, Connection>>>,
80
    host_uri: String,
81

            
82
    uldata: Arc<Mutex<Queue>>,
83
    dldata: Arc<Mutex<Queue>>,
84
    dldata_resp: Arc<Mutex<Queue>>,
85
    dldata_result: Arc<Mutex<Queue>>,
86

            
87
    status: Arc<Mutex<MgrStatus>>,
88
    handler: Arc<Mutex<Arc<dyn EventHandler>>>,
89
}
90

            
91
/// Event handler trait for the [`ApplicationMgr`].
92
#[async_trait]
93
pub trait EventHandler: Send + Sync {
94
    /// Fired when one of the manager's queues encounters a state change.
95
    async fn on_status_change(&self, mgr: &ApplicationMgr, status: MgrStatus);
96

            
97
    /// Fired when a [`UlData`] data is received.
98
    ///
99
    /// Return [`Err`] will NACK the data.
100
    /// The data may will be received again depending on the protocol (such as AMQP).
101
    async fn on_uldata(&self, mgr: &ApplicationMgr, data: Box<UlData>) -> Result<(), ()>;
102

            
103
    /// Fired when a [`DlDataResp`] data is received.
104
    ///
105
    /// Return [`Err`] will NACK the data.
106
    /// The data may will be received again depending on the protocol (such as AMQP).
107
    async fn on_dldata_resp(&self, mgr: &ApplicationMgr, data: Box<DlDataResp>) -> Result<(), ()>;
108

            
109
    /// Fired when a [`DlDataResult`] data is received.
110
    ///
111
    /// Return [`Err`] will NACK the data.
112
    /// The data may will be received again depending on the protocol (such as AMQP).
113
    async fn on_dldata_result(
114
        &self,
115
        mgr: &ApplicationMgr,
116
        data: Box<DlDataResult>,
117
    ) -> Result<(), ()>;
118
}
119

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

            
125
#[derive(Deserialize)]
126
struct UlDataInner {
127
    #[serde(rename = "dataId")]
128
    data_id: String,
129
    time: String,
130
    #[serde(rename = "pub")]
131
    publish: String,
132
    #[serde(rename = "deviceId")]
133
    device_id: String,
134
    #[serde(rename = "networkId")]
135
    network_id: String,
136
    #[serde(rename = "networkCode")]
137
    network_code: String,
138
    #[serde(rename = "networkAddr")]
139
    network_addr: String,
140
    #[serde(rename = "isPublic")]
141
    is_public: bool,
142
    data: String,
143
    extension: Option<Map<String, Value>>,
144
}
145

            
146
#[derive(Clone, Serialize)]
147
struct DlDataInner<'a> {
148
    #[serde(rename = "correlationId")]
149
    correlation_id: &'a String,
150
    #[serde(rename = "deviceId", skip_serializing_if = "Option::is_none")]
151
    device_id: Option<&'a String>,
152
    #[serde(rename = "networkCode", skip_serializing_if = "Option::is_none")]
153
    network_code: Option<&'a String>,
154
    #[serde(rename = "networkAddr", skip_serializing_if = "Option::is_none")]
155
    network_addr: Option<&'a String>,
156
    data: String,
157
    #[serde(skip_serializing_if = "Option::is_none")]
158
    extension: &'a Option<Map<String, Value>>,
159
}
160

            
161
const QUEUE_PREFIX: &'static str = "broker.application";
162
const ERR_PARAM_CORR_ID: &'static str = "the `correlation_id` must be a non-empty string";
163
const ERR_PARAM_DEV: &'static str = "one of `device_id` or [`network_code`, `network_addr`] pair must be provided with non-empty string";
164

            
165
impl ApplicationMgr {
166
    /// To create a manager instance.
167
68
    pub fn new(
168
68
        conn_pool: Arc<Mutex<HashMap<String, Connection>>>,
169
68
        host_uri: &Url,
170
68
        opts: Options,
171
68
        handler: Arc<dyn EventHandler>,
172
68
    ) -> Result<Self, String> {
173
68
        if opts.unit_id.len() == 0 {
174
8
            return Err("`unit_id` cannot be empty for application".to_string());
175
60
        }
176

            
177
60
        let conn = get_connection(&conn_pool, host_uri)?;
178

            
179
48
        let (uldata, dldata, dldata_resp, dldata_result, _) =
180
60
            new_data_queues(&conn, &opts, QUEUE_PREFIX, false)?;
181

            
182
48
        let mgr = ApplicationMgr {
183
48
            opts: Arc::new(opts),
184
48
            conn_pool,
185
48
            host_uri: host_uri.to_string(),
186
48
            uldata,
187
48
            dldata,
188
48
            dldata_resp: dldata_resp.unwrap(),
189
48
            dldata_result,
190
48
            status: Arc::new(Mutex::new(MgrStatus::NotReady)),
191
48
            handler: Arc::new(Mutex::new(handler)),
192
48
        };
193
48
        let mq_handler = Arc::new(MgrMqEventHandler { mgr: mgr.clone() });
194
48
        let mut q = { mgr.uldata.lock().unwrap().clone() };
195
48
        q.set_handler(mq_handler.clone());
196
48
        q.set_msg_handler(mq_handler.clone());
197
48
        if let Err(e) = q.connect() {
198
            return Err(e.to_string());
199
48
        }
200
48
        let mut q = { mgr.dldata.lock().unwrap().clone() };
201
48
        q.set_handler(mq_handler.clone());
202
48
        if let Err(e) = q.connect() {
203
            return Err(e.to_string());
204
48
        }
205
48
        let mut q = { mgr.dldata_resp.lock().unwrap().clone() };
206
48
        q.set_handler(mq_handler.clone());
207
48
        q.set_msg_handler(mq_handler.clone());
208
48
        if let Err(e) = q.connect() {
209
            return Err(e.to_string());
210
48
        }
211
48
        let mut q = { mgr.dldata_result.lock().unwrap().clone() };
212
48
        q.set_handler(mq_handler.clone());
213
48
        q.set_msg_handler(mq_handler.clone());
214
48
        if let Err(e) = q.connect() {
215
            return Err(e.to_string());
216
48
        }
217
48
        match conn {
218
24
            Connection::Amqp(_, counter) => {
219
24
                *counter.lock().unwrap() += 4;
220
24
            }
221
24
            Connection::Mqtt(_, counter) => {
222
24
                *counter.lock().unwrap() += 4;
223
24
            }
224
        }
225
48
        Ok(mgr)
226
68
    }
227

            
228
    /// The associated unit ID of the application.
229
4
    pub fn unit_id(&self) -> &str {
230
4
        self.opts.unit_id.as_str()
231
4
    }
232

            
233
    /// The associated unit code of the application.
234
4
    pub fn unit_code(&self) -> &str {
235
4
        self.opts.unit_code.as_str()
236
4
    }
237

            
238
    /// The application ID.
239
4
    pub fn id(&self) -> &str {
240
4
        self.opts.id.as_str()
241
4
    }
242

            
243
    /// The application code.
244
4
    pub fn name(&self) -> &str {
245
4
        self.opts.name.as_str()
246
4
    }
247

            
248
    /// Manager status.
249
216
    pub fn status(&self) -> MgrStatus {
250
216
        *self.status.lock().unwrap()
251
216
    }
252

            
253
    /// Detail status of each message queue. Please ignore `ctrl`.
254
4
    pub fn mq_status(&self) -> DataMqStatus {
255
4
        DataMqStatus {
256
4
            uldata: { self.uldata.lock().unwrap().status() },
257
4
            dldata: { self.dldata.lock().unwrap().status() },
258
4
            dldata_resp: { self.dldata_resp.lock().unwrap().status() },
259
4
            dldata_result: { self.dldata_result.lock().unwrap().status() },
260
4
            ctrl: QueueStatus::Closed,
261
4
        }
262
4
    }
263

            
264
    /// To close the manager queues.
265
    /// The underlying connection will be closed when there are no queues use it.
266
48
    pub async fn close(&self) -> Result<(), Box<dyn StdError + Send + Sync>> {
267
48
        let mut q = { self.uldata.lock().unwrap().clone() };
268
48
        q.close().await?;
269
48
        let mut q = { self.dldata.lock().unwrap().clone() };
270
48
        q.close().await?;
271
48
        let mut q = { self.dldata_resp.lock().unwrap().clone() };
272
48
        q.close().await?;
273
48
        let mut q = { self.dldata_result.lock().unwrap().clone() };
274
48
        q.close().await?;
275

            
276
48
        remove_connection(&self.conn_pool, &self.host_uri, 4).await
277
48
    }
278

            
279
    /// Send downlink data [`DlData`] to the broker.
280
24
    pub fn send_dldata(&self, data: &DlData) -> Result<(), Box<dyn StdError>> {
281
24
        if data.correlation_id.len() == 0 {
282
4
            let err = IoError::new(ErrorKind::InvalidInput, ERR_PARAM_CORR_ID.to_string());
283
4
            return Err(Box::new(err));
284
20
        }
285
20
        if data.device_id.is_none() {
286
12
            if data.network_code.is_none() || data.network_addr.is_none() {
287
4
                let err = IoError::new(ErrorKind::InvalidInput, ERR_PARAM_DEV.to_string());
288
4
                return Err(Box::new(err));
289
8
            } else if data.network_code.as_ref().unwrap().len() == 0
290
4
                || data.network_addr.as_ref().unwrap().len() == 0
291
            {
292
4
                let err = IoError::new(ErrorKind::InvalidInput, ERR_PARAM_DEV.to_string());
293
4
                return Err(Box::new(err));
294
4
            }
295
8
        } else if data.device_id.as_ref().unwrap().len() == 0 {
296
4
            let err = IoError::new(ErrorKind::InvalidInput, ERR_PARAM_DEV.to_string());
297
4
            return Err(Box::new(err));
298
4
        }
299

            
300
8
        let payload = serde_json::to_vec(&DlDataInner {
301
8
            correlation_id: &data.correlation_id,
302
8
            device_id: data.device_id.as_ref(),
303
8
            network_code: data.network_code.as_ref(),
304
8
            network_addr: data.network_addr.as_ref(),
305
8
            data: hex::encode(&data.data),
306
8
            extension: &data.extension,
307
8
        })?;
308
8
        let queue = { (*self.dldata.lock().unwrap()).clone() };
309
8
        task::spawn(async move {
310
8
            let _ = queue.send_msg(payload).await;
311
8
        });
312
8
        Ok(())
313
24
    }
314
}
315

            
316
#[async_trait]
317
impl QueueEventHandler for MgrMqEventHandler {
318
    async fn on_error(&self, _queue: Arc<dyn GmqQueue>, _err: Box<dyn StdError + Send + Sync>) {}
319

            
320
348
    async fn on_status(&self, _queue: Arc<dyn GmqQueue>, _status: QueueStatus) {
321
348
        let status = match { self.mgr.uldata.lock().unwrap().status() } == QueueStatus::Connected
322
136
            && { self.mgr.dldata.lock().unwrap().status() } == QueueStatus::Connected
323
116
            && { self.mgr.dldata_resp.lock().unwrap().status() } == QueueStatus::Connected
324
76
            && { self.mgr.dldata_result.lock().unwrap().status() } == QueueStatus::Connected
325
        {
326
310
            false => MgrStatus::NotReady,
327
38
            true => MgrStatus::Ready,
328
        };
329

            
330
348
        let mut changed = false;
331
348
        {
332
348
            let mut mutex = self.mgr.status.lock().unwrap();
333
348
            if *mutex != status {
334
76
                *mutex = status;
335
76
                changed = true;
336
272
            }
337
        }
338
348
        if changed {
339
76
            let handler = { self.mgr.handler.lock().unwrap().clone() };
340
76
            handler.on_status_change(&self.mgr, status).await;
341
272
        }
342
696
    }
343
}
344

            
345
#[async_trait]
346
impl MessageHandler for MgrMqEventHandler {
347
    // Validate and decode data.
348
66
    async fn on_message(&self, queue: Arc<dyn GmqQueue>, msg: Box<dyn Message>) {
349
        const _FN_NAME: &'static str = "ApplicationMgr.on_message";
350

            
351
66
        let queue_name = queue.name();
352
66
        if queue_name.cmp(self.mgr.uldata.lock().unwrap().name()) == Ordering::Equal {
353
30
            let data = match serde_json::from_slice::<UlDataInner>(msg.payload()) {
354
                Err(_) => {
355
4
                    let _ = msg.ack().await;
356
4
                    return;
357
                }
358
26
                Ok(data) => {
359
26
                    let data_bytes = match data.data.len() {
360
4
                        0 => vec![],
361
22
                        _ => match hex::decode(data.data.as_str()) {
362
                            Err(_) => {
363
4
                                let _ = msg.ack().await;
364
4
                                return;
365
                            }
366
18
                            Ok(data) => data,
367
                        },
368
                    };
369
22
                    let time = match DateTime::parse_from_rfc3339(data.time.as_str()) {
370
                        Err(_) => {
371
4
                            let _ = msg.ack().await;
372
4
                            return;
373
                        }
374
18
                        Ok(time) => time.into(),
375
                    };
376
18
                    let publish = match DateTime::parse_from_rfc3339(data.publish.as_str()) {
377
                        Err(_) => {
378
4
                            let _ = msg.ack().await;
379
4
                            return;
380
                        }
381
14
                        Ok(publish) => publish.into(),
382
14
                    };
383
14
                    UlData {
384
14
                        data_id: data.data_id,
385
14
                        time,
386
14
                        publish,
387
14
                        device_id: data.device_id,
388
14
                        network_id: data.network_id,
389
14
                        network_code: data.network_code,
390
14
                        network_addr: data.network_addr,
391
14
                        is_public: data.is_public,
392
14
                        data: data_bytes,
393
14
                        extension: data.extension,
394
14
                    }
395
14
                }
396
14
            };
397
14
            let handler = { self.mgr.handler.lock().unwrap().clone() };
398
14
            let _ = match handler.on_uldata(&self.mgr, Box::new(data)).await {
399
4
                Err(_) => msg.nack().await,
400
10
                Ok(_) => msg.ack().await,
401
            };
402
36
        } else if queue_name.cmp(self.mgr.dldata_resp.lock().unwrap().name()) == Ordering::Equal {
403
18
            let data = match serde_json::from_slice::<DlDataResp>(msg.payload()) {
404
                Err(_) => {
405
4
                    let _ = msg.ack().await;
406
4
                    return;
407
                }
408
14
                Ok(data) => data,
409
14
            };
410
14
            let handler = { self.mgr.handler.lock().unwrap().clone() };
411
14
            let _ = match handler.on_dldata_resp(&self.mgr, Box::new(data)).await {
412
4
                Err(_) => msg.nack().await,
413
10
                Ok(_) => msg.ack().await,
414
            };
415
18
        } else if queue_name.cmp(self.mgr.dldata_result.lock().unwrap().name()) == Ordering::Equal {
416
18
            let data = match serde_json::from_slice::<DlDataResult>(msg.payload()) {
417
                Err(_) => {
418
4
                    let _ = msg.ack().await;
419
4
                    return;
420
                }
421
14
                Ok(data) => data,
422
14
            };
423
14
            let handler = { self.mgr.handler.lock().unwrap().clone() };
424
14
            let _ = match handler.on_dldata_result(&self.mgr, Box::new(data)).await {
425
4
                Err(_) => msg.nack().await,
426
10
                Ok(_) => msg.ack().await,
427
            };
428
        }
429
132
    }
430
}