Lines
91.62 %
Functions
69.23 %
Branches
100 %
use std::{
collections::HashMap,
error::Error as StdError,
io::{Error as IoError, ErrorKind},
sync::{Arc, Mutex},
};
use axum::{response::IntoResponse, Router};
use reqwest;
use serde::{Deserialize, Serialize};
use url::Url;
use async_trait::async_trait;
use log::{error, info, warn};
use general_mq::{
queue::{EventHandler as QueueEventHandler, GmqQueue, Status},
Queue,
use sylvia_iot_corelib::{
constants::{CacheEngine, DbEngine},
http::{Json, Query},
use crate::{
libs::{
config::{self, Config},
mq::{self, application::ApplicationMgr, network::NetworkMgr, Connection},
},
models::{
self, Cache, CacheConnOptions, ConnOptions, DeviceOptions, DeviceRouteOptions, Model,
MongoDbOptions, NetworkRouteOptions, 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 `/broker`, the APIs are
/// - `http://host:port/broker/api/v1/unit/xxx`
/// - `http://host:port/broker/api/v1/application/xxx`
pub scope_path: &'static str,
/// The scopes for accessing APIs.
pub api_scopes: HashMap<String, Vec<String>>,
/// The database model.
pub model: Arc<dyn Model>,
/// The database cache.
pub cache: Option<Arc<dyn Cache>>,
/// The sylvia-iot-auth base API path with host.
/// For example, `http://localhost:1080/auth`.
pub auth_base: String,
pub amqp_prefetch: u16,
pub amqp_persistent: bool,
pub mqtt_shared_prefix: String,
/// The client for internal HTTP requests.
pub client: reqwest::Client,
/// Queue connections. Key is uri.
pub mq_conns: Arc<Mutex<HashMap<String, Connection>>>,
/// Application managers. Key is `[unit-code].[application-code]`.
pub application_mgrs: Arc<Mutex<HashMap<String, ApplicationMgr>>>,
/// Network managers. Key is `[unit-code].[network-code]`. Unit code `_` means public network.
pub network_mgrs: Arc<Mutex<HashMap<String, NetworkMgr>>>,
/// Control channel receivers. Key is function such as `application`, `network`, ....
pub ctrl_receivers: Arc<Mutex<HashMap<String, Queue>>>,
/// Control channel senders.
pub ctrl_senders: CtrlSenders,
/// Data channel sender.
pub data_sender: Option<Queue>,
}
pub struct CtrlSenders {
pub unit: Arc<Mutex<Queue>>,
pub application: Arc<Mutex<Queue>>,
pub network: Arc<Mutex<Queue>>,
pub device: Arc<Mutex<Queue>>,
pub device_route: Arc<Mutex<Queue>>,
pub network_route: Arc<Mutex<Queue>>,
/// The sylvia-iot module specific error codes in addition to standard
/// [`sylvia_iot_corelib::err::ErrResp`].
pub struct ErrReq;
struct DataSenderHandler;
/// Query parameters for `GET /version`
#[derive(Deserialize)]
pub struct GetVersionQuery {
q: Option<String>,
#[derive(Serialize)]
struct GetVersionRes<'a> {
data: GetVersionResData<'a>,
struct GetVersionResData<'a> {
name: &'a str,
version: &'a str,
const SERV_NAME: &'static str = env!("CARGO_PKG_NAME");
const SERV_VER: &'static str = env!("CARGO_PKG_VERSION");
impl ErrReq {
pub const APPLICATION_EXIST: (u16, &'static str) = (400, "err_broker_application_exist");
pub const APPLICATION_NOT_EXIST: (u16, &'static str) =
(400, "err_broker_application_not_exist");
pub const DEVICE_NOT_EXIST: (u16, &'static str) = (400, "err_broker_device_not_exist");
pub const MEMBER_NOT_EXIST: (u16, &'static str) = (400, "err_broker_member_not_exist");
pub const NETWORK_ADDR_EXIST: (u16, &'static str) = (400, "err_broker_network_addr_exist");
pub const NETWORK_EXIST: (u16, &'static str) = (400, "err_broker_network_exist");
pub const NETWORK_NOT_EXIST: (u16, &'static str) = (400, "err_broker_network_not_exist");
pub const OWNER_NOT_EXIST: (u16, &'static str) = (400, "err_broker_owner_not_exist");
pub const ROUTE_EXIST: (u16, &'static str) = (400, "err_broker_route_exist");
pub const UNIT_EXIST: (u16, &'static str) = (400, "err_broker_unit_exist");
pub const UNIT_NOT_EXIST: (u16, &'static str) = (400, "err_broker_unit_not_exist");
pub const UNIT_NOT_MATCH: (u16, &'static str) = (400, "err_broker_unit_not_match");
#[async_trait]
impl QueueEventHandler for DataSenderHandler {
async fn on_error(&self, queue: Arc<dyn GmqQueue>, err: Box<dyn StdError + Send + Sync>) {
const FN_NAME: &'static str = "DataSenderHandler::on_error";
let queue_name = queue.name();
error!("[{}] {} error: {}", FN_NAME, queue_name, err);
async fn on_status(&self, queue: Arc<dyn GmqQueue>, status: Status) {
const FN_NAME: &'static str = "DataSenderHandler::on_status";
match status {
Status::Connected => info!("[{}] {} connected", queue_name, FN_NAME),
_ => warn!("[{}] {} status to {:?}", FN_NAME, queue_name, status),
/// 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 cache_opts = match conf.cache.as_ref().unwrap().engine.as_ref() {
None => None,
Some(engine) => match engine.as_str() {
CacheEngine::MEMORY => {
let conf = conf.cache.as_ref().unwrap().memory.as_ref().unwrap();
Some(CacheConnOptions::Memory {
device: DeviceOptions {
uldata_size: conf.device.unwrap(),
device_route: DeviceRouteOptions {
uldata_size: conf.device_route.unwrap(),
dldata_size: conf.device_route.unwrap(),
dldata_pub_size: conf.device_route.unwrap(),
network_route: NetworkRouteOptions {
_ => None,
let mq_conf = conf.mq.as_ref().unwrap();
let model = models::new(&db_opts).await?;
let cache = match cache_opts {
Some(opts) => Some(models::new_cache(&opts, &model).await?),
let auth_base = conf.auth.as_ref().unwrap().clone();
let mq_conns = Arc::new(Mutex::new(HashMap::new()));
let ch_conf = conf.mq_channels.as_ref().unwrap();
let ctrl_senders = new_ctrl_senders(&mq_conns, &ch_conf, cache.clone())?;
let data_sender = match ch_conf.data.as_ref() {
Some(conf) => match conf.url.as_ref() {
Some(_) => Some(new_data_sender(&mq_conns, conf)?),
let state = State {
scope_path: match scope_path.len() {
0 => "/",
_ => scope_path,
api_scopes: conf.api_scopes.as_ref().unwrap().clone(),
model,
cache,
auth_base,
amqp_prefetch: mq_conf.prefetch.unwrap(),
amqp_persistent: mq_conf.persistent.unwrap(),
mqtt_shared_prefix: mq_conf.shared_prefix.as_ref().unwrap().to_string(),
client: reqwest::Client::new(),
mq_conns,
application_mgrs: Arc::new(Mutex::new(HashMap::new())),
network_mgrs: Arc::new(Mutex::new(HashMap::new())),
ctrl_receivers: Arc::new(Mutex::new(HashMap::new())),
ctrl_senders,
data_sender,
let (r1, r2, r3, r4, r5, r6) = tokio::join!(
v1::unit::init(&state, &ch_conf.unit.as_ref().unwrap()),
v1::application::init(&state, &ch_conf.application.as_ref().unwrap()),
v1::network::init(&state, &ch_conf.network.as_ref().unwrap()),
v1::device::init(&state, &ch_conf.device.as_ref().unwrap()),
v1::device_route::init(&state, &ch_conf.device_route.as_ref().unwrap()),
v1::network_route::init(&state, &ch_conf.network_route.as_ref().unwrap())
);
r1?;
r2?;
r3?;
r4?;
r5?;
r6?;
Ok(state)
/// To register service URIs in the specified root path.
pub fn new_service(state: &State) -> Router {
Router::new().nest(
&state.scope_path,
Router::new()
.merge(v1::unit::new_service("/api/v1/unit", state))
.merge(v1::application::new_service("/api/v1/application", state))
.merge(v1::network::new_service("/api/v1/network", state))
.merge(v1::device::new_service("/api/v1/device", state))
.merge(v1::device_route::new_service("/api/v1/device-route", state))
.merge(v1::network_route::new_service(
"/api/v1/network-route",
state,
))
.merge(v1::dldata_buffer::new_service(
"/api/v1/dldata-buffer",
)),
)
pub fn new_ctrl_senders(
mq_conns: &Arc<Mutex<HashMap<String, Connection>>>,
ch_conf: &config::MqChannels,
cache: Option<Arc<dyn Cache>>,
) -> Result<CtrlSenders, Box<dyn StdError>> {
let unit_ctrl_cfg = ch_conf.unit.as_ref().unwrap();
let app_ctrl_cfg = ch_conf.application.as_ref().unwrap();
let net_ctrl_cfg = ch_conf.network.as_ref().unwrap();
let dev_ctrl_cfg = ch_conf.device.as_ref().unwrap();
let devr_ctrl_cfg = ch_conf.device_route.as_ref().unwrap();
let netr_ctrl_cfg = ch_conf.network_route.as_ref().unwrap();
Ok(CtrlSenders {
unit: v1::unit::new_ctrl_sender(mq_conns, unit_ctrl_cfg)?,
application: v1::application::new_ctrl_sender(mq_conns, app_ctrl_cfg)?,
network: v1::network::new_ctrl_sender(mq_conns, net_ctrl_cfg, cache.clone())?,
device: v1::device::new_ctrl_sender(mq_conns, dev_ctrl_cfg, cache.clone())?,
device_route: v1::device_route::new_ctrl_sender(mq_conns, devr_ctrl_cfg, cache.clone())?,
network_route: v1::network_route::new_ctrl_sender(mq_conns, netr_ctrl_cfg, cache.clone())?,
/// Create data channel sender queue.
pub fn new_data_sender(
conn_pool: &Arc<Mutex<HashMap<String, Connection>>>,
config: &config::BrokerData,
) -> Result<Queue, Box<dyn StdError>> {
let url = match config.url.as_ref() {
None => {
return Err(Box::new(IoError::new(
ErrorKind::InvalidInput,
"empty control url",
)))
Some(url) => match Url::parse(url.as_str()) {
Err(e) => return Err(Box::new(e)),
Ok(url) => url,
let persistent = match config.persistent {
None => false,
Some(persistent) => persistent,
match mq::data::new(conn_pool, &url, persistent, Arc::new(DataSenderHandler {})) {
Err(e) => Err(Box::new(IoError::new(ErrorKind::InvalidInput, e))),
Ok(q) => Ok(q),
pub async fn get_version(Query(query): Query<GetVersionQuery>) -> impl IntoResponse {
if let Some(q) = query.q.as_ref() {
match q.as_str() {
"name" => return SERV_NAME.into_response(),
"version" => return SERV_VER.into_response(),
_ => (),
Json(GetVersionRes {
data: GetVersionResData {
name: SERV_NAME,
version: SERV_VER,
.into_response()