Method stream_device_events

chirpstack/src/api/internal.rs:843–907

(
        &self,
        request: Request<api::StreamDeviceEventsRequest>,
    )

    type StreamDeviceEventsStream = DropReceiver<Result<api::LogItem, Status>>;

    async fn stream_device_events(
        &self,
        request: Request<api::StreamDeviceEventsRequest>,
    ) -> Result<Response<Self::StreamDeviceEventsStream>, Status> {
        let req = request.get_ref();
        let dev_eui = EUI64::from_str(&req.dev_eui).map_err(|e| e.status())?;

        self.validator
            .validate(
                request.extensions(),
                validator::ValidateDeviceAccess::new(validator::Flag::Read, dev_eui),
            )
            .await?;

        let key = redis_key(format!("device:{{{}}}:stream:event", req.dev_eui));
        let (redis_tx, mut redis_rx) = mpsc::channel(1);
        let (stream_tx, stream_rx) = mpsc::channel(1);

        let mut eventlog_future = Box::pin(stream::event::get_event_logs(key, 10, redis_tx));
        let (drop_receiver, mut close_rx) = DropReceiver::new(ReceiverStream::new(stream_rx));

        tokio::spawn(async move {
            loop {
                tokio::select! {
                    // detect client disconnect
                    _ = close_rx.recv() => {
                        debug!("Client disconnected");
                        redis_rx.close();
                        break;
                    },
                    // detect get_event_logs function return
                    res = &mut eventlog_future => {
                        match res {
                            Ok(_) => {
                                trace!("get_event_logs returned");
                            },
                            Err(e) => {
                                error!("Reading event-log returned error: {}", e);
                                stream_tx.send(Err(e.status())).await.unwrap();
                            },
                        }
                        break;
                    }
                    // detect stream message
                    msg = redis_rx.recv() => {
                        match msg {
                            None => {
                                trace!("Redis Stream channel has been closed");
                                break;
                            },
                            Some(msg) => {
                                trace!("Message received from Redis Stream channel");
                                if stream_tx.send(Ok(msg)).await.is_err() {
                                    error!("Sending message to gRPC channel error");
                                    break;
                                };
                            },
                        }