rufish 0.4.0

An asynchronous Redfish client library for BMC/server management in Rust.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

//! Server-Sent Events (SSE) support for real-time Redfish event streaming.

use std::pin::Pin;
use std::task::{Context, Poll};

use futures_core::Stream;
use serde::{Deserialize, Serialize};
use tokio_stream::StreamExt;

/// A parsed SSE event from the Redfish EventService.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RedfishEvent {
    /// Event type (e.g. "Event").
    pub event_type: Option<String>,
    /// Event ID.
    pub id: Option<String>,
    /// Raw data payload (JSON string).
    pub data: String,
    /// Parsed event body, if valid JSON.
    pub body: Option<serde_json::Value>,
}

/// A stream of Redfish SSE events.
pub struct EventStream {
    inner: Pin<Box<dyn Stream<Item = Result<RedfishEvent, crate::error::RedfishError>> + Send>>,
}

impl Stream for EventStream {
    type Item = Result<RedfishEvent, crate::error::RedfishError>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.inner.as_mut().poll_next(cx)
    }
}

impl EventStream {
    pub(crate) fn new(response: reqwest::Response) -> Self {
        let stream = async_stream(response);
        Self { inner: Box::pin(stream) }
    }
}

fn async_stream(
    response: reqwest::Response,
) -> impl Stream<Item = Result<RedfishEvent, crate::error::RedfishError>> {
    use tokio_stream::wrappers::ReceiverStream;

    let (tx, rx) = tokio::sync::mpsc::channel(64);

    tokio::spawn(async move {
        let mut bytes_stream = response.bytes_stream();
        let mut buffer = String::new();
        let mut event_type: Option<String> = None;
        let mut event_id: Option<String> = None;
        let mut data_lines: Vec<String> = Vec::new();

        while let Some(chunk) = bytes_stream.next().await {
            let chunk = match chunk {
                Ok(c) => c,
                Err(e) => {
                    let _ = tx.send(Err(crate::error::RedfishError::Http(e))).await;
                    break;
                }
            };

            buffer.push_str(&String::from_utf8_lossy(&chunk));

            while let Some(pos) = buffer.find('\n') {
                let line = buffer[..pos].trim_end_matches('\r').to_string();
                buffer = buffer[pos + 1..].to_string();

                if line.is_empty() {
                    // Empty line = end of event
                    if !data_lines.is_empty() {
                        let data = data_lines.join("\n");
                        let body = serde_json::from_str(&data).ok();
                        let event = RedfishEvent {
                            event_type: event_type.take(),
                            id: event_id.take(),
                            data,
                            body,
                        };
                        if tx.send(Ok(event)).await.is_err() {
                            return;
                        }
                    }
                    data_lines.clear();
                } else if let Some(value) = line.strip_prefix("data:") {
                    data_lines.push(value.trim_start().to_string());
                } else if let Some(value) = line.strip_prefix("event:") {
                    event_type = Some(value.trim_start().to_string());
                } else if let Some(value) = line.strip_prefix("id:") {
                    event_id = Some(value.trim_start().to_string());
                }
            }
        }
    });

    ReceiverStream::new(rx)
}