Function message_callback

chirpstack/src/integration/mqtt.rs:399–443

(
    application_id: String,
    dev_eui: String,
    command: String,
    json: bool,
    p: Publish,
)

}

async fn message_callback(
    application_id: String,
    dev_eui: String,
    command: String,
    json: bool,
    p: Publish,
) {
    let topic = String::from_utf8_lossy(&p.topic);

    info!(topic = %topic, qos = ?p.qos, "Command received for device");

    let err = || -> Result<()> {
        match command.as_ref() {
            "down" => {
                let cmd: integration::DownlinkCommand = match json {
                    true => serde_json::from_slice(&p.payload)?,
                    false => integration::DownlinkCommand::decode(&mut Cursor::new(&p.payload))?,
                };
                if dev_eui != cmd.dev_eui {
                    return Err(anyhow!(
                        "Payload dev_eui {} does not match topic dev_eui {}",
                        cmd.dev_eui,
                        dev_eui
                    ));
                }
                tokio::spawn(super::handle_down_command(application_id, cmd));
            }
            _ => {
                return Err(anyhow!("Unknown command type"));
            }
        }

        Ok(())
    }()
    .err();

    if err.is_some() {
        warn!(
            topic = %topic,
            qos = ?p.qos,
            "Processing command error: {}",
            err.as_ref().unwrap()
        );
    }
}

#[cfg(all(test, feature = "test-integration-mqtt"))]
pub mod test {