Function handle_stream

chirpstack/src/stream/frame.rs:281–376

(
    stream_id: &str,
    channel: &mpsc::Sender<api::LogItem>,
    k: &str,
    v: &redis::Value,
)

}

async fn handle_stream(
    stream_id: &str,
    channel: &mpsc::Sender<api::LogItem>,
    k: &str,
    v: &redis::Value,
) -> Result<()> {
    match k {
        "up" => {
            trace!(key = %k, id = %stream_id, "Frame-log received from stream");
            if let redis::Value::BulkString(b) = v {
                let pl = stream::UplinkFrameLog::decode(&mut Cursor::new(b))?;
                let mut phy = lrwn::PhyPayload::from_slice(&pl.phy_payload)?;
                if pl.plaintext_f_opts
                    && let Err(e) = phy.decode_f_opts_to_mac_commands()
                {
                    warn!(error = %e.full(), "Decode f_opts to mac-commands error");
                }
                if pl.plaintext_frm_payload
                    && let Err(e) = phy.decode_frm_payload()
                {
                    warn!(error = %e.full(), "Decode frm_payload error");
                }

                let pl = api::LogItem {
                    id: stream_id.to_string(),
                    time: pl.time.as_ref().map(|t| prost_types::Timestamp {
                        seconds: t.seconds,
                        nanos: t.nanos,
                    }),
                    description: pl.f_type().into(),
                    body: json!({
                        "phy_payload": phy,
                        "tx_info": pl.tx_info,
                        "rx_info": pl.rx_info,
                    })
                    .to_string(),
                    properties: [
                        ("DevAddr".to_string(), pl.dev_addr),
                        ("DevEUI".to_string(), pl.dev_eui),
                    ]
                    .iter()
                    .cloned()
                    .collect(),
                };

                channel.send(pl).await?;
            }
        }
        "down" => {
            trace!(key = %k, id = %stream_id, "frame-log received from stream");
            if let redis::Value::BulkString(b) = v {
                let pl = stream::DownlinkFrameLog::decode(&mut Cursor::new(b))?;
                let mut phy = lrwn::PhyPayload::from_slice(&pl.phy_payload)?;
                if pl.plaintext_f_opts
                    && let Err(e) = phy.decode_f_opts_to_mac_commands()
                {
                    warn!(error = %e.full(), "Decode f_opts to mac-commands error");
                }