Method events_stream

crates/openshell-driver-podman/src/client.rs:633–720

Start streaming container events filtered by label. Events are sent to the returned receiver. The background task runs until the receiver is dropped.

(
        &self,
        label_filter: &str,
    )

    /// Events are sent to the returned receiver. The background task runs
    /// until the receiver is dropped.
    pub async fn events_stream(
        &self,
        label_filter: &str,
    ) -> Result<mpsc::Receiver<Result<PodmanEvent, PodmanApiError>>, PodmanApiError> {
        let filters = serde_json::json!({
            "label": [label_filter],
            "type": ["container"],
        });
        let encoded = url_encode(&filters.to_string());
        let path =
            format!("http://localhost/{API_VERSION}/libpod/events?stream=true&filters={encoded}");

        let mut sender = self.connect().await?;

        let req = Request::builder()
            .method(hyper::Method::GET)
            .uri(&path)
            .header("Host", "localhost")
            .body(Full::new(Bytes::new()))
            .map_err(|e| PodmanApiError::Connection(e.to_string()))?;

        let response = tokio::time::timeout(API_TIMEOUT, sender.send_request(req))
            .await
            .map_err(|_| PodmanApiError::Timeout(API_TIMEOUT))?
            .map_err(|e| PodmanApiError::Connection(e.to_string()))?;

        if !response.status().is_success() {
            return Err(PodmanApiError::Api {
                status: response.status().as_u16(),
                message: "events stream request failed".to_string(),
            });
        }

        let (tx, rx) = mpsc::channel(256);
        let body = response.into_body();

        tokio::spawn(async move {
            let mut buffer = Vec::new();
            let mut body = body;

            loop {
                use hyper::body::Body;

                let frame =
                    match std::future::poll_fn(|cx| Pin::new(&mut body).poll_frame(cx)).await {
                        Some(Ok(frame)) => frame,
                        Some(Err(e)) => {
                            let _ = tx
                                .send(Err(PodmanApiError::Connection(e.to_string())))
                                .await;
                            break;
                        }
                        None => break,
                    };

                if let Some(data) = frame.data_ref() {
                    buffer.extend_from_slice(data);
                }