Lines
95.02 %
Functions
90.91 %
Branches
100 %
use std::{
cmp::Ordering,
collections::HashMap,
error::Error as StdError,
sync::{Arc, Mutex},
};
use async_trait::async_trait;
use hex;
use log::{error, warn};
use serde::{Deserialize, Serialize};
use serde_json::{self, Map, Value};
use tokio::task;
use url::Url;
use general_mq::{
queue::{
EventHandler as QueueEventHandler, GmqQueue, Message, MessageHandler, Status as QueueStatus,
},
Queue,
use sylvia_iot_corelib::{err, strings};
use super::{
get_connection, new_data_queues, remove_connection, Connection, MgrMqStatus, MgrStatus, Options,
/// Uplink data from broker to application.
#[derive(Serialize)]
pub struct UlData {
#[serde(rename = "dataId")]
pub data_id: String,
pub time: String,
#[serde(rename = "pub")]
pub publish: String,
#[serde(rename = "deviceId")]
pub device_id: String,
#[serde(rename = "networkId")]
pub network_id: String,
#[serde(rename = "networkCode")]
pub network_code: String,
#[serde(rename = "networkAddr")]
pub network_addr: String,
#[serde(rename = "isPublic")]
pub is_public: bool,
pub profile: String,
pub data: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub extension: Option<Map<String, Value>>,
}
/// Downlink data from application to broker.
#[derive(Deserialize)]
pub struct DlData {
#[serde(rename = "correlationId")]
pub correlation_id: String,
pub device_id: Option<String>,
pub network_code: Option<String>,
pub network_addr: Option<String>,
/// Downlink data response for [`DlData`].
#[derive(Default, Serialize)]
pub struct DlDataResp {
#[serde(rename = "dataId", skip_serializing_if = "Option::is_none")]
pub data_id: Option<String>,
pub error: Option<String>,
pub message: Option<String>,
/// Downlink data result when processing or completing data transfer to the device.
pub struct DlDataResult {
pub status: i32,
/// The manager for application queues.
#[derive(Clone)]
pub struct ApplicationMgr {
opts: Arc<Options>,
// Information for delete connection automatically.
conn_pool: Arc<Mutex<HashMap<String, Connection>>>,
host_uri: String,
uldata: Arc<Mutex<Queue>>,
dldata: Arc<Mutex<Queue>>,
dldata_resp: Arc<Mutex<Queue>>,
dldata_result: Arc<Mutex<Queue>>,
status: Arc<Mutex<MgrStatus>>,
handler: Arc<Mutex<Arc<dyn EventHandler>>>,
/// Event handler trait for the [`ApplicationMgr`].
#[async_trait]
pub trait EventHandler: Send + Sync {
async fn on_status_change(&self, mgr: &ApplicationMgr, status: MgrStatus);
async fn on_dldata(
&self,
mgr: &ApplicationMgr,
data: Box<DlData>,
) -> Result<Box<DlDataResp>, ()>;
/// The event handler for [`general_mq::queue::GmqQueue`].
struct MgrMqEventHandler {
mgr: ApplicationMgr,
const QUEUE_PREFIX: &'static str = "broker.application";
impl ApplicationMgr {
/// To create a manager instance.
pub fn new(
host_uri: &Url,
opts: Options,
handler: Arc<dyn EventHandler>,
) -> Result<Self, String> {
if opts.unit_id.len() == 0 {
return Err("`unit_id` cannot be empty for application".to_string());
let conn = get_connection(&conn_pool, host_uri)?;
let (uldata, dldata, dldata_resp, dldata_result) =
new_data_queues(&conn, &opts, QUEUE_PREFIX, false)?;
let mgr = ApplicationMgr {
opts: Arc::new(opts),
conn_pool,
host_uri: host_uri.to_string(),
uldata,
dldata,
dldata_resp: dldata_resp.unwrap(),
dldata_result,
status: Arc::new(Mutex::new(MgrStatus::NotReady)),
handler: Arc::new(Mutex::new(handler)),
let mq_handler = Arc::new(MgrMqEventHandler { mgr: mgr.clone() });
let mut q = { mgr.uldata.lock().unwrap().clone() };
q.set_handler(mq_handler.clone());
if let Err(e) = q.connect() {
return Err(e.to_string());
let mut q = { mgr.dldata.lock().unwrap().clone() };
q.set_msg_handler(mq_handler.clone());
let mut q = { mgr.dldata_resp.lock().unwrap().clone() };
let mut q = { mgr.dldata_result.lock().unwrap().clone() };
match conn {
Connection::Amqp(_, counter) => {
*counter.lock().unwrap() += 4;
Connection::Mqtt(_, counter) => {
Ok(mgr)
/// The associated unit ID of the application.
pub fn unit_id(&self) -> &str {
self.opts.unit_id.as_str()
/// The associated unit code of the application.
pub fn unit_code(&self) -> &str {
self.opts.unit_code.as_str()
/// The application ID.
pub fn id(&self) -> &str {
self.opts.id.as_str()
/// The application code.
pub fn name(&self) -> &str {
self.opts.name.as_str()
/// Manager status.
pub fn status(&self) -> MgrStatus {
*self.status.lock().unwrap()
/// Detail status of each message queue.
pub fn mq_status(&self) -> MgrMqStatus {
MgrMqStatus {
uldata: { self.uldata.lock().unwrap().status() },
dldata: { self.dldata.lock().unwrap().status() },
dldata_resp: { self.dldata_resp.lock().unwrap().status() },
dldata_result: { self.dldata_result.lock().unwrap().status() },
ctrl: QueueStatus::Closed,
/// To close the manager queues.
pub async fn close(&self) -> Result<(), Box<dyn StdError + Send + Sync>> {
let mut q = { self.uldata.lock().unwrap().clone() };
q.close().await?;
let mut q = { self.dldata.lock().unwrap().clone() };
let mut q = { self.dldata_resp.lock().unwrap().clone() };
let mut q = { self.dldata_result.lock().unwrap().clone() };
remove_connection(&self.conn_pool, &self.host_uri, 4).await
/// Send uplink data to the application.
pub fn send_uldata(&self, data: &UlData) -> Result<(), Box<dyn StdError>> {
let payload = serde_json::to_vec(data)?;
let queue = { (*self.uldata.lock().unwrap()).clone() };
task::spawn(async move {
let _ = queue.send_msg(payload).await;
});
Ok(())
/// Send downlink response for a downlink data to the application.
pub async fn send_dldata_resp(
data: &DlDataResp,
) -> Result<(), Box<dyn StdError + Send + Sync>> {
let queue = { (*self.dldata_resp.lock().unwrap()).clone() };
queue.send_msg(payload).await
/// Send the downlink data process result to the application.
pub async fn send_dldata_result(
data: &DlDataResult,
let queue = { (*self.dldata_result.lock().unwrap()).clone() };
impl QueueEventHandler for MgrMqEventHandler {
async fn on_error(&self, _queue: Arc<dyn GmqQueue>, _err: Box<dyn StdError + Send + Sync>) {}
async fn on_status(&self, _queue: Arc<dyn GmqQueue>, _status: QueueStatus) {
let status = match { self.mgr.uldata.lock().unwrap().status() } == QueueStatus::Connected
&& { self.mgr.dldata.lock().unwrap().status() } == QueueStatus::Connected
&& { self.mgr.dldata_resp.lock().unwrap().status() } == QueueStatus::Connected
&& { self.mgr.dldata_result.lock().unwrap().status() } == QueueStatus::Connected
{
false => MgrStatus::NotReady,
true => MgrStatus::Ready,
let mut changed = false;
let mut mutex = self.mgr.status.lock().unwrap();
if *mutex != status {
*mutex = status;
changed = true;
if changed {
let handler = { self.mgr.handler.lock().unwrap().clone() };
handler.on_status_change(&self.mgr, status).await;
impl MessageHandler for MgrMqEventHandler {
// Validate and decode data.
async fn on_message(&self, queue: Arc<dyn GmqQueue>, msg: Box<dyn Message>) {
const FN_NAME: &'static str = "ApplicationMgr.on_message";
let queue_name = queue.name();
if queue_name.cmp(self.mgr.dldata.lock().unwrap().name()) == Ordering::Equal {
let data = match parse_dldata_msg(msg.payload()) {
Err(resp) => {
warn!("[{}] invalid format from {}", FN_NAME, queue_name);
if let Err(e) = msg.ack().await {
error!("[{}] ACK message error: {}", FN_NAME, e);
if let Err(e) = self.mgr.send_dldata_resp(&resp).await {
error!("[{}] send response error: {}", FN_NAME, e);
return;
Ok(data) => data,
match handler.on_dldata(&self.mgr, Box::new(data)).await {
Err(_) => {
if let Err(e) = msg.nack().await {
error!("[{}] NACK message error: {}", FN_NAME, e);
Ok(resp) => {
if let Err(e) = self.mgr.send_dldata_resp(resp.as_ref()).await {
/// Parses downlink data from the application and responds error for wrong format data.
fn parse_dldata_msg(msg: &[u8]) -> Result<DlData, DlDataResp> {
let mut data = match serde_json::from_slice::<DlData>(msg) {
return Err(DlDataResp {
correlation_id: "".to_string(),
error: Some(err::E_PARAM.to_string()),
message: Some("invalid format".to_string()),
..Default::default()
if data.correlation_id.len() == 0 {
correlation_id: data.correlation_id.clone(),
message: Some("invalid `correlationId`".to_string()),
match data.device_id.as_ref() {
None => {
match data.network_code.as_ref() {
message: Some("missing `networkCode`".to_string()),
Some(code) => {
let code = code.to_lowercase();
match strings::is_code(code.as_str()) {
false => {
message: Some("invalid `networkCode`".to_string()),
true => {
data.network_code = Some(code);
()
match data.network_addr.as_ref() {
message: Some("missing `networkAddr`".to_string()),
Some(addr) => match addr.len() {
0 => {
message: Some("invalid `networkAddr`".to_string()),
_ => {
data.network_addr = Some(addr.to_lowercase());
Some(device_id) => match device_id.len() {
message: Some("invalid `deviceId`".to_string()),
_ => (),
if data.data.len() > 0 {
if let Err(_) = hex::decode(data.data.as_str()) {
message: Some("invalid `data`".to_string()),
data.data = data.data.to_lowercase();
Ok(data)