Method start

chirpstack-integration/src/lib.rs:160–279

(&self)

    }

    async fn start(&self) -> Result<()> {
        info!("Getting Redis connection");
        let mut redis_conn = self.redis_client.get_async_connection().await?;

        let key = format!("{}device:stream:event", self.key_prefix);

        // Try to create the consumer group. This will fail in case the consumer group already exists.
        let _: usize = match redis::cmd("XGROUP")
            .arg("CREATE")
            .arg(&key)
            .arg(&self.consumer_group)
            .arg(0)
            .arg("MKSTREAM")
            .query_async(&mut redis_conn)
            .await
        {
            Ok(v) => v,
            Err(e) => {
                warn!(error = %e, "Could not create Redis consumer group, ignore this error if the group already exists");
                0
            }
        };

        loop {
            let srr: redis::streams::StreamReadReply = redis::cmd("XREADGROUP")
                .arg("GROUP")
                .arg(&self.consumer_group)
                .arg(&self.consumer_name)
                .arg("COUNT")
                .arg(10)
                .arg("BLOCK")
                .arg(1000)
                .arg("STREAMS")
                .arg(&key)
                .arg(">")
                .query_async(&mut redis_conn)
                .await?;

            for stream_key in &srr.keys {
                for stream_id in &stream_key.ids {
                    let _: () = redis::cmd("XACK")
                        .arg(&key)
                        .arg(&self.consumer_group)
                        .arg(&stream_id.id)
                        .query_async(&mut redis_conn)
                        .await?;

                    for (k, v) in &stream_id.map {
                        let res = || -> Result<()> {
                            info!(key = %k, "Event received from Redis stream");
                            match k.as_ref() {
                                "up" => {
                                    if let redis::Value::BulkString(b) = v {
                                        let pl = integration_pb::UplinkEvent::decode(
                                            &mut Cursor::new(b),
                                        )?;
                                        tokio::spawn(uplink_event(pl));
                                    }