#[derive(Deserialize)]

#[derive(Deserialize, Serialize)]

#[derive(Deserialize, Serialize)]

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.application.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.application().list(&opts, cursor).await?;

        for item in list.iter() {

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

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

            let opts = MgrOptions {

                unit_id: item.unit_id.clone(),

                unit_code: item.unit_code.clone(),

                id: item.application_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(),

                network_mgrs: state.network_mgrs.clone(),

                data_sender: state.data_sender.clone(),

            };

            let mgr =

                match ApplicationMgr::new(state.mq_conns.clone(), &url, opts, Arc::new(handler)) {

                    Ok(mgr) => mgr,

                };

            {

                state

                    .application_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,

) -> 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 {}),

        Arc::new(CtrlSenderHandler {}),

    ) {

        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_application(

    State(state): State<AppState>,

    Extension(token_info): Extension<GetTokenInfoData>,

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

) -> 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 unit_id = body.data.unit_id.as_str();

    if unit_id.len() == 0 {

        return Err(ErrResp::ErrParam(Some(

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

        )));

    }

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

            return Err(ErrResp::Custom(

                ErrReq::UNIT_NOT_EXIST.0,

                ErrReq::UNIT_NOT_EXIST.1,

                None,

            ))

        Some(unit) => unit.code,

    };

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

        return Err(ErrResp::Custom(

            ErrReq::APPLICATION_EXIST.0,

            ErrReq::APPLICATION_EXIST.1,

            None,

        ));

    }

    let now = Utc::now();

    let application = Application {

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

        code,

        unit_id: unit_id.to_string(),

        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.application().add(&application).await {

    }

    add_manager(

        FN_NAME,

        &state,

        &host_uri,

        unit_id,

        unit_code.as_str(),

        application.application_id.as_str(),

        application.code.as_str(),

    )

    Ok(Json(response::PostApplication {

        data: response::PostApplicationData {

            application_id: application.application_id,

        },

    }))

}

pub async fn get_application_count(

    State(state): State<AppState>,

    Extension(token_info): Extension<GetTokenInfoData>,

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

) -> 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 => 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(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.application().count(&cond).await {

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

            data: response::GetCountData { count },

        })),

}

pub async fn get_application_list(

    State(state): State<AppState>,

    Extension(token_info): Extension<GetTokenInfoData>,

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

) -> 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 => 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(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.application().list(&opts, None).await {

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

            None => match query.format {

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

                    return Ok(Json(response::GetApplicationList {

                        data: application_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 => None,

                _ => 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 application_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.application().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_application(

    State(state): State<AppState>,

    Extension(token_info): Extension<GetTokenInfoData>,

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

) -> impl IntoResponse {

    let user_id = token_info.user_id.as_str();

    let roles = &token_info.roles;

    let application_id = param.application_id.as_str();

    match check_application(FN_NAME, application_id, user_id, false, roles, &state).await? {

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

        Some(application) => Ok(Json(response::GetApplication {

            data: application_transform(&application),

        })),

}

pub async fn patch_application(

    State(state): State<AppState>,

    Extension(token_info): Extension<GetTokenInfoData>,

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

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

) -> impl IntoResponse {

    let user_id = token_info.user_id.as_str();

    let roles = &token_info.roles;

    let application_id = param.application_id.as_str();

    let application =

        match check_application(FN_NAME, application_id, user_id, true, roles, &state).await? {

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

            Some(application) => application,

    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(application.host_uri.as_str()) {

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

            should_add_mgr = true;

    }

    let cond = UpdateQueryCond { application_id };

    if let Err(e) = state.model.application().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();

            add_manager(

                FN_NAME,

                &state,

                &uri,

                application.unit_id.as_str(),

                application.unit_code.as_str(),

                application.application_id.as_str(),

                application.code.as_str(),

            )

    }

    Ok(StatusCode::NO_CONTENT)

}

pub async fn delete_application(

    State(state): State<AppState>,

    Extension(token_info): Extension<GetTokenInfoData>,

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

) -> impl IntoResponse {

    let user_id = token_info.user_id.as_str();

    let roles = &token_info.roles;

    let application_id = param.application_id.as_str();

    let application =

        match check_application(FN_NAME, application_id, user_id, true, roles, &state).await {

            Ok(application) => match application {

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

                Some(application) => application,

            },

        };

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

    del_application_rsc(FN_NAME, application_id, &state).await?;

    send_del_ctrl_message(FN_NAME, application, &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::PatchApplicationData,

) -> 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: &str,

    code: &str,

    state: &AppState,

) -> Result<bool, ErrResp> {

    let cond = QueryCond {

        unit_id: Some(unit_id),

        code: Some(code),

        ..Default::default()

    };

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

        Ok(application) => match application {

            None => Ok(false),

            Some(_) => Ok(true),

}

fn application_list_transform(list: &Vec<Application>) -> Vec<response::GetApplicationData> {

    let mut ret = vec![];

    for application in list.iter() {

        ret.push(application_transform(&application));

    }

    ret

}

fn application_list_transform_bytes(

    list: &Vec<Application>,

    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(&application_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 application_transform(application: &Application) -> response::GetApplicationData {

    response::GetApplicationData {

        application_id: application.application_id.clone(),

        code: application.code.clone(),

        unit_id: application.unit_id.clone(),

        unit_code: application.unit_code.clone(),

        created_at: time_str(&application.created_at),

        modified_at: time_str(&application.modified_at),

        host_uri: application.host_uri.clone(),

        name: application.name.clone(),

        info: application.info.clone(),

    }

}

async fn del_application_rsc(

    fn_name: &str,

    application_id: &str,

    state: &AppState,

) -> Result<(), ErrResp> {

    let cond = network_route::QueryCond {

        application_id: Some(application_id),

        ..Default::default()

    };

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

    }

    let cond = device_route::QueryCond {

        application_id: Some(application_id),

        ..Default::default()

    };

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

    }

    let cond = dldata_buffer::QueryCond {

        application_id: Some(application_id),

        ..Default::default()

    };

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

    }

    let cond = QueryCond {

        application_id: Some(application_id),

        ..Default::default()

    };

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

    }

    Ok(())

}

async fn send_del_ctrl_message(

    fn_name: &str,

    application: Application,

    state: &AppState,

) -> Result<(), ErrResp> {

    if state.cache.is_some() {

        let msg = SendCtrlMsg::DelApplication {

            operation: CtrlMsgOp::DEL_APPLICATION.to_string(),

            new: CtrlDelApplication {

                unit_id: application.unit_id,

                unit_code: application.unit_code,

                application_id: application.application_id,

                application_code: application.code,

            },

        };

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

            Ok(payload) => payload,

        };

        let ctrl_sender = { state.ctrl_senders.application.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.application.lock().unwrap().clone() };

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

    }

    Ok(())

}

async fn delete_manager(

    fn_name: &str,

    state: &AppState,

    application: &Application,

) -> Result<(), ErrResp> {

    let key = gen_mgr_key(application.unit_code.as_str(), application.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.application.lock().unwrap().clone() };

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

    }

    Ok(())

}

    async fn get_device_route(

        &self,

        mgr: &ApplicationMgr,

        data: &Box<DlData>,

    ) -> Result<device_route::DeviceRouteCacheDlData, Box<DlDataResp>> {

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

            match data.device_id.as_ref() {

                    let cond = device_route::GetCacheQueryCond {

                        unit_code: mgr.unit_code(),

                        network_code: data.network_code.as_ref().unwrap().as_str(),

                        network_addr: data.network_addr.as_ref().unwrap().as_str(),

                    };

                    match cache.device_route().get_dldata(&cond).await {

                        Ok(route) => match route {

                                warn!(

                                return Err(Box::new(DlDataResp {

                                    correlation_id: data.correlation_id.clone(),

                                    error: Some(ErrReq::DEVICE_NOT_EXIST.1.to_string()),

                                    ..Default::default()

                                }));

                            Some(route) => return Ok(route),

                Some(device_id) => {

                    let cond = device_route::GetCachePubQueryCond {

                        unit_id: mgr.unit_id(),

                        device_id: device_id.as_str(),

                    };

                    match cache.device_route().get_dldata_pub(&cond).await {

                        Ok(route) => match route {

                            Some(route) => return Ok(route),

        }

        let cond = match data.device_id.as_ref() {

            None => device::QueryCond {

                device: Some(device::QueryOneCond {

                    unit_code: Some(mgr.unit_code()),

                    network_code: data.network_code.as_ref().unwrap().as_str(),

                    network_addr: data.network_addr.as_ref().unwrap().as_str(),

                }),

                ..Default::default()

            },

            Some(device_id) => device::QueryCond {

                unit_id: Some(mgr.unit_id()),

                device_id: Some(device_id.as_str()),

                ..Default::default()

            },

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

            Ok(device) => match device {

                    warn!(

                    return Err(Box::new(DlDataResp {

                        correlation_id: data.correlation_id.clone(),

                        error: Some(ErrReq::DEVICE_NOT_EXIST.1.to_string()),

                        ..Default::default()

                    }));

                Some(device) => device,

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

            None => "",

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

        Ok(device_route::DeviceRouteCacheDlData {

            net_mgr_key: gen_mgr_key(unit_code, device.network_code.as_str()),

            network_id: device.network_id,

            network_addr: device.network_addr,

            device_id: device.device_id,

            profile: device.profile,

        })

    }

    async fn send_application_dldata_msg(

        &self,

        proc: &DateTime<Utc>,

        data_id: &str,

        unit_id: &str,

        profile: &str,

        data: &Box<DlData>,

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

        if let Some(sender) = self.data_sender.as_ref() {

            let msg = SendDataMsg {

                kind: DataMsgKind::APP_DLDATA.to_string(),

                data: SendDataKind::AppDlData {

                    data_id: data_id.to_string(),

                    proc: time_str(proc),

                    status: DEF_DLDATA_STATUS,

                    unit_id: unit_id.to_string(),

                    device_id: data.device_id.clone(),

                    network_code: data.network_code.clone(),

                    network_addr: data.network_addr.clone(),

                    profile: profile.to_string(),

                    data: data.data.clone(),

                    extension: data.extension.clone(),

                },

            };

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

                Ok(payload) => payload,

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

            }

        }

        Ok(())