pub async fn init(state: &AppState, ctrl_conf: &CfgCtrl) -> Result<(), Box<dyn StdError>> {

    let q = new_ctrl_receiver(state, ctrl_conf)?;

    {

        state

            .ctrl_receivers

            .lock()

            .unwrap()

            .insert(CTRL_QUEUE_NAME.to_string(), q.clone());

    }

    let ctrl_sender = { state.ctrl_senders.network.lock().unwrap().clone() };

    for _ in 0..500 {

        if ctrl_sender.status() == Status::Connected && q.status() == Status::Connected {

            break;

        }

        time::sleep(Duration::from_millis(10)).await;

    if ctrl_sender.status() != Status::Connected {

    }

    if q.status() != Status::Connected {

    }

    let cond = ListQueryCond {

        ..Default::default()

    };

    let opts = ListOptions {

        cond: &cond,

        offset: None,

        limit: None,

        sort: None,

        cursor_max: Some(LIST_CURSOR_MAX),

    };

    let mut list;

    let mut cursor = None;

        (list, cursor) = state.model.network().list(&opts, cursor).await?;

        for item in list.iter() {

            let url = Url::parse(item.host_uri.as_str())?;

            let unit_id = match item.unit_id.as_ref() {

                Some(unit_id) => unit_id.as_str(),

            let unit_code = match item.unit_code.as_ref() {

                Some(unit_code) => unit_code.as_str(),

            let key = gen_mgr_key(unit_code, item.code.as_str());

            let opts = MgrOptions {

                unit_id: unit_id.to_string(),

                unit_code: unit_code.to_string(),

                id: item.network_id.clone(),

                name: item.code.clone(),

                prefetch: Some(state.amqp_prefetch),

                persistent: state.amqp_persistent,

                shared_prefix: Some(state.mqtt_shared_prefix.clone()),

            };

            let handler = MgrHandler {

                model: state.model.clone(),

                cache: state.cache.clone(),

                application_mgrs: state.application_mgrs.clone(),

                data_sender: state.data_sender.clone(),

            };

            let mgr = match NetworkMgr::new(state.mq_conns.clone(), &url, opts, Arc::new(handler)) {

                Ok(mgr) => mgr,

            };

            {

                state.network_mgrs.lock().unwrap().insert(key.clone(), mgr);

            }

        if cursor.is_none() {

            break;

    Ok(())

}

pub fn new_ctrl_sender(

    conn_pool: &Arc<Mutex<HashMap<String, Connection>>>,

    config: &CfgCtrl,

    cache: Option<Arc<dyn Cache>>,

) -> Result<Arc<Mutex<Queue>>, Box<dyn StdError>> {

    let url = match config.url.as_ref() {

        Some(url) => match Url::parse(url.as_str()) {

            Ok(url) => url,

        },

    };

    match mq::control::new(

        conn_pool.clone(),

        &url,

        config.prefetch,

        CTRL_QUEUE_NAME,

        false,

        Arc::new(CtrlSenderHandler {

            cache: cache.clone(),

        }),

        Arc::new(CtrlSenderHandler { cache }),

    ) {

        Ok(q) => Ok(Arc::new(Mutex::new(q))),

}

pub fn new_ctrl_receiver(state: &AppState, config: &CfgCtrl) -> Result<Queue, Box<dyn StdError>> {

    let url = match config.url.as_ref() {

        Some(url) => match Url::parse(url.as_str()) {

            Ok(url) => url,

        },

    };

    let handler = Arc::new(CtrlReceiverHandler {

        model: state.model.clone(),

        cache: state.cache.clone(),

        mq_conns: state.mq_conns.clone(),

        application_mgrs: state.application_mgrs.clone(),

        network_mgrs: state.network_mgrs.clone(),

        data_sender: state.data_sender.clone(),

    });

    match mq::control::new(

        state.mq_conns.clone(),

        &url,

        config.prefetch,

        CTRL_QUEUE_NAME,

        true,

        handler.clone(),

        handler,

    ) {

        Ok(q) => Ok(q),

}

pub async fn post_network(

    State(state): State<AppState>,

    Extension(token_info): Extension<GetTokenInfoData>,

    Json(body): Json<request::PostNetworkBody>,

) -> impl IntoResponse {

    let user_id = token_info.user_id.as_str();

    let roles = &token_info.roles;

    let code = body.data.code.to_lowercase();

    let host_uri = body.data.host_uri.as_str();

    if !strings::is_code(code.as_str()) {

        return Err(ErrResp::ErrParam(Some(

            "`code` must be [A-Za-z0-9]{1}[A-Za-z0-9-_]*".to_string(),

        )));

    }

    let host_uri = match Url::parse(host_uri) {

        Err(_) => return Err(ErrResp::ErrParam(Some("invalid `hostUri`".to_string()))),

        Ok(uri) => match mq::SUPPORT_SCHEMES.contains(&uri.scheme()) {

                return Err(ErrResp::ErrParam(Some(

                    "unsupport `hostUri` scheme".to_string(),

                )))

            true => uri,

    if let Some(info) = body.data.info.as_ref() {

        for (k, _) in info.iter() {

            if k.len() == 0 {

                return Err(ErrResp::ErrParam(Some(

                    "`info` key must not be empty".to_string(),

                )));

            }

    }

    let mut unit_code: Option<String> = None;

    let unit_id = match body.data.unit_id.as_ref() {

            if !Role::is_role(roles, Role::ADMIN) && !Role::is_role(roles, Role::MANAGER) {

                return Err(ErrResp::ErrParam(Some("missing `unitId`".to_string())));

            }

            None

        Some(unit_id) => {

            if unit_id.len() == 0 {

                return Err(ErrResp::ErrParam(Some(

                    "`unitId` must with at least one character".to_string(),

                )));

            }

            match check_unit(FN_NAME, user_id, roles, unit_id.as_str(), true, &state).await? {

                    return Err(ErrResp::Custom(

                        ErrReq::UNIT_NOT_EXIST.0,

                        ErrReq::UNIT_NOT_EXIST.1,

                        None,

                    ))

                Some(unit) => {

                    unit_code = Some(unit.code);

                }

            }

            Some(unit_id.clone())

    if check_code(FN_NAME, unit_id.as_ref(), code.as_str(), &state).await? {

        return Err(ErrResp::Custom(

            ErrReq::NETWORK_EXIST.0,

            ErrReq::NETWORK_EXIST.1,

            None,

        ));

    }

    let now = Utc::now();

    let network = Network {

        network_id: strings::random_id(&now, ID_RAND_LEN),

        code: code.clone(),

        unit_id: unit_id.clone(),

        unit_code: unit_code.clone(),

        created_at: now,

        modified_at: now,

        host_uri: host_uri.to_string(),

        name: match body.data.name.as_ref() {

            None => "".to_string(),

            Some(name) => name.clone(),

        info: match body.data.info.as_ref() {

            None => Map::new(),

            Some(info) => info.clone(),

    if let Err(e) = state.model.network().add(&network).await {

    }

    let unit_id = match unit_id.as_ref() {

        None => "",

        Some(id) => id.as_str(),

    let unit_code = match unit_code.as_ref() {

        None => "",

        Some(code) => code.as_str(),

    add_manager(

        FN_NAME,

        &state,

        &host_uri,

        unit_id,

        unit_code,

        network.network_id.as_str(),

        code.as_str(),

    )

    .await?;

    Ok(Json(response::PostNetwork {

        data: response::PostNetworkData {

            network_id: network.network_id,

        },

    }))

}

pub async fn get_network_count(

    State(state): State<AppState>,

    Extension(token_info): Extension<GetTokenInfoData>,

    Query(query): Query<request::GetNetworkCountQuery>,

) -> impl IntoResponse {

    let user_id = token_info.user_id.as_str();

    let roles = &token_info.roles;

    if !Role::is_role(roles, Role::ADMIN) && !Role::is_role(roles, Role::MANAGER) {

        match query.unit.as_ref() {

            None => return Err(ErrResp::ErrParam(Some("missing `unit`".to_string()))),

            Some(unit_id) => {

                if unit_id.len() == 0 {

                    return Err(ErrResp::ErrParam(Some("missing `unit`".to_string())));

                }

    }

    let unit_cond = match query.unit.as_ref() {

        None => None,

        Some(unit_id) => match unit_id.len() {

            0 => Some(None),

                match check_unit(FN_NAME, user_id, roles, unit_id.as_str(), false, &state).await? {

                        return Err(ErrResp::Custom(

                            ErrReq::UNIT_NOT_EXIST.0,

                            ErrReq::UNIT_NOT_EXIST.1,

                            None,

                        ))

                    Some(_) => Some(Some(unit_id.as_str())),

    let mut code_cond = None;

    let mut code_contains_cond = None;

    if let Some(code) = query.code.as_ref() {

        if code.len() > 0 {

            code_cond = Some(code.as_str());

        }

    }

    if code_cond.is_none() {

        if let Some(contains) = query.contains.as_ref() {

            if contains.len() > 0 {

                code_contains_cond = Some(contains.as_str());

            }

        }

    }

    let cond = ListQueryCond {

        unit_id: unit_cond,

        code: code_cond,

        code_contains: code_contains_cond,

        ..Default::default()

    };

    match state.model.network().count(&cond).await {

        Ok(count) => Ok(Json(response::GetNetworkCount {

            data: response::GetCountData { count },

        })),

}

pub async fn get_network_list(

    State(state): State<AppState>,

    Extension(token_info): Extension<GetTokenInfoData>,

    Query(query): Query<request::GetNetworkListQuery>,

) -> impl IntoResponse {

    let user_id = token_info.user_id.as_str();

    let roles = &token_info.roles;

    if !Role::is_role(roles, Role::ADMIN) && !Role::is_role(roles, Role::MANAGER) {

        match query.unit.as_ref() {

            None => return Err(ErrResp::ErrParam(Some("missing `unit`".to_string()))),

            Some(unit_id) => {

                if unit_id.len() == 0 {

                    return Err(ErrResp::ErrParam(Some("missing `unit`".to_string())));

                }

    }

    let unit_cond = match query.unit.as_ref() {

        None => None,

        Some(unit_id) => match unit_id.len() {

            0 => Some(None),

                match check_unit(FN_NAME, user_id, roles, unit_id.as_str(), false, &state).await? {

                        return Err(ErrResp::Custom(

                            ErrReq::UNIT_NOT_EXIST.0,

                            ErrReq::UNIT_NOT_EXIST.1,

                            None,

                        ))

                    Some(_) => Some(Some(unit_id.as_str())),

    let mut code_cond = None;

    let mut code_contains_cond = None;

    if let Some(code) = query.code.as_ref() {

        if code.len() > 0 {

            code_cond = Some(code.as_str());

        }

    }

    if code_cond.is_none() {

        if let Some(contains) = query.contains.as_ref() {

            if contains.len() > 0 {

                code_contains_cond = Some(contains.as_str());

            }

        }

    }

    let cond = ListQueryCond {

        unit_id: unit_cond,

        code: code_cond,

        code_contains: code_contains_cond,

        ..Default::default()

    };

    let sort_cond = get_sort_cond(&query.sort)?;

    let opts = ListOptions {

        cond: &cond,

        offset: query.offset,

        limit: match query.limit {

            None => Some(LIST_LIMIT_DEFAULT),

            Some(limit) => match limit {

                0 => None,

                _ => Some(limit),

        sort: Some(sort_cond.as_slice()),

        cursor_max: Some(LIST_CURSOR_MAX),

    let (list, cursor) = match state.model.network().list(&opts, None).await {

        Ok((list, cursor)) => match cursor {

            None => match query.format {

                    return Ok(Json(network_list_transform(&list)).into_response())

                    return Ok(Json(response::GetNetworkList {

                        data: network_list_transform(&list),

                    })

                    .into_response())

            Some(_) => (list, cursor),

        },

    };

    let body = Body::from_stream(async_stream::stream! {

        let unit_cond = match query.unit.as_ref() {

            None => None,

            Some(unit_id) => match unit_id.len() {

                0 => Some(None),

                _ => Some(Some(unit_id.as_str())),

            },

        };

        let mut code_contains_cond = None;

        if let Some(contains) = query.contains.as_ref() {

            if contains.len() > 0 {

                code_contains_cond = Some(contains.as_str());

            }

        }

        let cond = ListQueryCond {

            unit_id: unit_cond,

            code_contains: code_contains_cond,

            ..Default::default()

        };

        let opts = ListOptions {

            cond: &cond,

            offset: query.offset,

            limit: match query.limit {

                None => Some(LIST_LIMIT_DEFAULT),

                Some(limit) => match limit {

                    0 => None,

                    _ => Some(limit),

                },

            },

            sort: Some(sort_cond.as_slice()),

            cursor_max: Some(LIST_CURSOR_MAX),

        };

        let mut list = list;

        let mut cursor = cursor;

        let mut is_first = true;

        loop {

            yield network_list_transform_bytes(&list, is_first, cursor.is_none(), query.format.as_ref());

            is_first = false;

            if cursor.is_none() {

                break;

            }

            let (_list, _cursor) = match state.model.network().list(&opts, cursor).await {

                Err(_) => break,

                Ok((list, cursor)) => (list, cursor),

            };

            list = _list;

            cursor = _cursor;

        }

    });

    Ok(([(header::CONTENT_TYPE, ContentType::JSON)], body).into_response())

}

pub async fn get_network(

    State(state): State<AppState>,

    Extension(token_info): Extension<GetTokenInfoData>,

    Path(param): Path<request::NetworkIdPath>,

) -> impl IntoResponse {

    let user_id = token_info.user_id.as_str();

    let roles = &token_info.roles;

    let network_id = param.network_id.as_str();

    match check_network(FN_NAME, network_id, user_id, false, roles, &state).await? {

        None => Err(ErrResp::ErrNotFound(None)),

        Some(network) => Ok(Json(response::GetNetwork {

            data: network_transform(&network),

        })),

}

pub async fn patch_network(

    State(state): State<AppState>,

    Extension(token_info): Extension<GetTokenInfoData>,

    Path(param): Path<request::NetworkIdPath>,

    Json(mut body): Json<request::PatchNetworkBody>,

) -> impl IntoResponse {

    let user_id = token_info.user_id.as_str();

    let roles = &token_info.roles;

    let network_id = param.network_id.as_str();

    let network = match check_network(FN_NAME, network_id, user_id, true, roles, &state).await? {

        None => return Err(ErrResp::ErrNotFound(None)),

        Some(network) => network,

    let updates = get_updates(&mut body.data).await?;

    let mut should_add_mgr = false;

    if let Some(host_uri) = updates.host_uri {

        let uri = Url::parse(host_uri).unwrap();

        if !uri.as_str().eq(network.host_uri.as_str()) {

            delete_manager(FN_NAME, &state, &network).await?;

            should_add_mgr = true;

    }

    let cond = UpdateQueryCond { network_id };

    if let Err(e) = state.model.network().update(&cond, &updates).await {

    }

    // Add new manager.

    if should_add_mgr {

        if let Some(host_uri) = updates.host_uri {

            let uri = Url::parse(host_uri).unwrap();

            let unit_id = match network.unit_id.as_ref() {

                None => "",

                Some(id) => id.as_str(),

            let unit_code = match network.unit_code.as_ref() {

                None => "",

                Some(code) => code.as_str(),

            add_manager(

                FN_NAME,

                &state,

                &uri,

                unit_id,

                unit_code,

                network.network_id.as_str(),

                network.code.as_str(),

            )

            .await?;

    }

    Ok(StatusCode::NO_CONTENT)

}

pub async fn delete_network(

    State(state): State<AppState>,

    Extension(token_info): Extension<GetTokenInfoData>,

    Path(param): Path<request::NetworkIdPath>,

) -> impl IntoResponse {

    let user_id = token_info.user_id.as_str();

    let roles = &token_info.roles;

    let network_id = param.network_id.as_str();

    let network = match check_network(FN_NAME, network_id, user_id, true, roles, &state).await {

        Ok(network) => match network {

            None => return Ok(StatusCode::NO_CONTENT),

            Some(network) => network,

        },

    };

    delete_manager(FN_NAME, &state, &network).await?;

    del_network_rsc(FN_NAME, network_id, &state).await?;

    send_del_ctrl_message(FN_NAME, network, &state).await?;

    Ok(StatusCode::NO_CONTENT)

}

fn get_sort_cond(sort_args: &Option<String>) -> Result<Vec<SortCond>, ErrResp> {

    match sort_args.as_ref() {

        None => Ok(vec![SortCond {

            key: SortKey::Code,

            asc: true,

        }]),

        Some(args) => {

            let mut args = args.split(",");

            let mut sort_cond = vec![];

            while let Some(arg) = args.next() {

                let mut cond = arg.split(":");

                let key = match cond.next() {

                    Some(field) => match field {

                        "code" => SortKey::Code,

                        "created" => SortKey::CreatedAt,

                        "modified" => SortKey::ModifiedAt,

                        "name" => SortKey::Name,

                            return Err(ErrResp::ErrParam(Some(format!(

                                "invalid sort key {}",

                                field

                            ))))

                let asc = match cond.next() {

                    None => return Err(ErrResp::ErrParam(Some("wrong sort argument".to_string()))),

                    Some(asc) => match asc {

                        "asc" => true,

                        "desc" => false,

                            return Err(ErrResp::ErrParam(Some(format!(

                                "invalid sort asc {}",

                                asc

                            ))))

                if cond.next().is_some() {

                    return Err(ErrResp::ErrParam(Some(

                        "invalid sort condition".to_string(),

                    )));

                }

                sort_cond.push(SortCond { key, asc });

            Ok(sort_cond)

}

async fn get_updates<'a>(body: &'a mut request::PatchNetworkData) -> Result<Updates<'a>, ErrResp> {

    let mut updates = Updates {

        ..Default::default()

    };

    let mut count = 0;

    if let Some(host_uri) = body.host_uri.as_ref() {

        match Url::parse(host_uri) {

            Err(_) => return Err(ErrResp::ErrParam(Some("invalid `hostUri`".to_string()))),

            Ok(uri) => {

                if !mq::SUPPORT_SCHEMES.contains(&uri.scheme()) {

                    return Err(ErrResp::ErrParam(Some(

                        "unsupport `hostUri` scheme".to_string(),

                    )));

                }

                body.host_uri = Some(uri.to_string()); // change host name case.

    }

    if let Some(host_uri) = body.host_uri.as_ref() {

        updates.host_uri = Some(host_uri.as_str());

        count += 1;

    }

    if let Some(name) = body.name.as_ref() {

        updates.name = Some(name.as_str());

        count += 1;

    }

    if let Some(info) = body.info.as_ref() {

        for (k, _) in info.iter() {

            if k.len() == 0 {

                return Err(ErrResp::ErrParam(Some(

                    "`info` key must not be empty".to_string(),

                )));

            }

        updates.info = Some(info);

        count += 1;

    }

    if count == 0 {

        return Err(ErrResp::ErrParam(Some(

            "at least one parameter".to_string(),

        )));

    }

    updates.modified_at = Some(Utc::now());

    Ok(updates)

}

async fn check_code(

    fn_name: &str,

    unit_id: Option<&String>,

    code: &str,

    state: &AppState,

) -> Result<bool, ErrResp> {

    let cond = QueryCond {

        unit_id: match unit_id {

            None => Some(None),

            Some(unit_id) => Some(Some(unit_id.as_str())),

        code: Some(code),

        ..Default::default()

    };

    match state.model.network().get(&cond).await {

        Ok(network) => match network {

            None => Ok(false),

            Some(_) => Ok(true),

}

fn network_list_transform(list: &Vec<Network>) -> Vec<response::GetNetworkData> {

    let mut ret = vec![];

    for network in list.iter() {

        ret.push(network_transform(&network));

    }

    ret

}

fn network_list_transform_bytes(

    list: &Vec<Network>,

    with_start: bool,

    with_end: bool,

    format: Option<&request::ListFormat>,

) -> Result<Bytes, Box<dyn StdError + Send + Sync>> {

    let mut build_str = match with_start {

        false => "".to_string(),

        true => match format {

            Some(request::ListFormat::Array) => "[".to_string(),

            _ => "{\"data\":[".to_string(),

    let mut is_first = with_start;

    for item in list {

        if is_first {

            is_first = false;

        } else {

            build_str.push(',');

        }

        let json_str = match serde_json::to_string(&network_transform(item)) {

            Ok(str) => str,

        };

        build_str += json_str.as_str();

    if with_end {

        build_str += match format {

            Some(request::ListFormat::Array) => "]",

            _ => "]}",

    }

    Ok(Bytes::copy_from_slice(build_str.as_str().as_bytes()))

}

fn network_transform(network: &Network) -> response::GetNetworkData {

    response::GetNetworkData {

        network_id: network.network_id.clone(),

        code: network.code.clone(),

        unit_id: network.unit_id.clone(),

        unit_code: network.unit_code.clone(),

        created_at: time_str(&network.created_at),

        modified_at: time_str(&network.modified_at),

        host_uri: network.host_uri.clone(),

        name: network.name.clone(),

        info: network.info.clone(),

    }

}

async fn del_network_rsc(fn_name: &str, network_id: &str, state: &AppState) -> Result<(), ErrResp> {

    let cond = network_route::QueryCond {

        network_id: Some(network_id),

        ..Default::default()

    };

    if let Err(e) = state.model.network_route().del(&cond).await {

    }

    let cond = device_route::QueryCond {

        network_id: Some(network_id),

        ..Default::default()

    };

    if let Err(e) = state.model.device_route().del(&cond).await {

    }

    let cond = dldata_buffer::QueryCond {

        network_id: Some(network_id),

        ..Default::default()

    };

    if let Err(e) = state.model.dldata_buffer().del(&cond).await {

    }

    let cond = device::QueryCond {

        network_id: Some(network_id),

        ..Default::default()

    };

    if let Err(e) = state.model.device().del(&cond).await {

    }

    let cond = QueryCond {

        network_id: Some(network_id),

        ..Default::default()

    };

    if let Err(e) = state.model.network().del(&cond).await {

    }

    Ok(())

}

async fn send_del_ctrl_message(

    fn_name: &str,

    network: Network,

    state: &AppState,

) -> Result<(), ErrResp> {

    if state.cache.is_some() {

        let msg = SendCtrlMsg::DelNetwork {

            operation: CtrlMsgOp::DEL_NETWORK.to_string(),

            new: CtrlDelNetwork {

                unit_id: network.unit_id,

                unit_code: network.unit_code,

                network_id: network.network_id,

                network_code: network.code,

            },

        };

        let payload = match serde_json::to_vec(&msg) {

            Ok(payload) => payload,

        };

        let ctrl_sender = { state.ctrl_senders.network.lock().unwrap().clone() };

        if let Err(e) = ctrl_sender.send_msg(payload).await {

        }

    }

    Ok(())

}

async fn add_manager(

    fn_name: &str,

    state: &AppState,

    host_uri: &Url,

    unit_id: &str,

    unit_code: &str,

    id: &str,

    name: &str,

) -> Result<(), ErrResp> {

    let opts = MgrOptions {

        unit_id: unit_id.to_string(),

        unit_code: unit_code.to_string(),

        id: id.to_string(),

        name: name.to_string(),

        prefetch: Some(state.amqp_prefetch),

        persistent: state.amqp_persistent,

        shared_prefix: Some(state.mqtt_shared_prefix.clone()),

    };

    let msg = SendCtrlMsg::AddManager {

        operation: CtrlMsgOp::ADD_MANAGER.to_string(),

        new: CtrlAddManager {

            host_uri: host_uri.to_string(),

            mgr_options: opts,

        },

    };

    let payload = match serde_json::to_vec(&msg) {

        Ok(payload) => payload,

    };

    let ctrl_sender = { state.ctrl_senders.network.lock().unwrap().clone() };

    if let Err(e) = ctrl_sender.send_msg(payload).await {

    }

    Ok(())

}

async fn delete_manager(fn_name: &str, state: &AppState, network: &Network) -> Result<(), ErrResp> {

    let key = match network.unit_code.as_ref() {

        None => gen_mgr_key("", network.code.as_str()),

        Some(unit_code) => gen_mgr_key(unit_code.as_str(), network.code.as_str()),

    let msg = SendCtrlMsg::DelManager {

        operation: CtrlMsgOp::DEL_MANAGER.to_string(),

        new: key.clone(),

    };

    let payload = match serde_json::to_vec(&msg) {

        Ok(payload) => payload,

    };

    let ctrl_sender = { state.ctrl_senders.network.lock().unwrap().clone() };

    if let Err(e) = ctrl_sender.send_msg(payload).await {

    }

    Ok(())

}

    async fn validate_device(

        &self,

        mgr: &NetworkMgr,

        data: &Box<UlData>,

    ) -> Result<Option<DeviceCacheItem>, Box<dyn StdError>> {

        let dev_cond = device::QueryOneCond {

            unit_code: match mgr.unit_code().len() {

                0 => None,

                _ => Some(mgr.unit_code()),

            network_code: mgr.name(),

            network_addr: data.network_addr.as_str(),

        };

        match self.cache.as_ref() {

                let cond = device::QueryCond {

                    device: Some(dev_cond.clone()),

                    ..Default::default()

                };

                match self.model.device().get(&cond).await {

                    Ok(device) => match device {

                            warn!(

                            Ok(None)

                        Some(device) => Ok(Some(DeviceCacheItem {

                            device_id: device.device_id,

                            profile: device.profile,

                        })),

            Some(cache) => {

                let cond = device::GetCacheQueryCond::CodeAddr(dev_cond.clone());

                match cache.device().get(&cond).await {

                    Ok(device) => match device {

                            warn!(

                            Ok(None)

                        Some(device) => Ok(Some(device)),

    }

    async fn send_by_device_route(

        &self,

        netmgr_unit_code: Option<String>,

        app_data: &mut ApplicationUlData,

        sent_mgrs: &mut Vec<String>,

    ) -> Result<(), ()> {

        if let Some(cache) = self.cache.as_ref() {

            let route = match cache

                .device_route()

                .get_uldata(&app_data.device_id.as_str())

                .await

                Ok(route) => match route {

                    None => return Ok(()),

                    Some(route) => route,

            for key in route.app_mgr_keys.iter() {

                if sent_mgrs.contains(key) {

                }

                let mgr = {

                    match self.application_mgrs.lock().unwrap().get(key) {

                        Some(mgr) => mgr.clone(),

                    }

                };

                let now = Utc::now();

                app_data.data_id = strings::random_id(&now, DATA_ID_RAND_LEN);

                if let Err(e) = mgr.send_uldata(&app_data) {

                }

                self.send_application_uldata_msg(

                    &now,

                    netmgr_unit_code.clone(),

                    mgr.unit_id().to_string(),