rc_core/
watch.rs

1//! Live object notification streaming contracts.
2
3use std::pin::Pin;
4
5use async_trait::async_trait;
6use futures::Stream;
7use jiff::Timestamp;
8use serde::{Deserialize, Serialize};
9
10use crate::Result;
11
12/// A boxed stream of live notification frames.
13pub type WatchStream = Pin<Box<dyn Stream<Item = Result<WatchFrame>> + Send + 'static>>;
14
15/// Filters and connection settings for one live notification request.
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub struct WatchRequest {
18    /// Optional bucket. `None` watches the whole alias endpoint.
19    pub bucket: Option<String>,
20    /// S3 event names sent as repeated `events` query parameters.
21    pub events: Vec<String>,
22    /// Optional decoded object-key prefix filter.
23    pub prefix: Option<String>,
24    /// Optional decoded object-key suffix filter.
25    pub suffix: Option<String>,
26    /// RustFS keepalive interval in seconds.
27    pub ping_seconds: u64,
28}
29
30/// One decoded frame from a live notification stream.
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub enum WatchFrame {
33    /// A server keepalive that must not be shown as an object event.
34    KeepAlive,
35    /// One object notification record.
36    Event(Box<WatchEvent>),
37}
38
39/// Optional origin metadata supplied by RustFS for an object event.
40#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
41pub struct WatchSource {
42    #[serde(skip_serializing_if = "Option::is_none")]
43    pub event_source: Option<String>,
44    #[serde(skip_serializing_if = "Option::is_none")]
45    pub region: Option<String>,
46    #[serde(skip_serializing_if = "Option::is_none")]
47    pub principal_id: Option<String>,
48    #[serde(skip_serializing_if = "Option::is_none")]
49    pub host: Option<String>,
50    #[serde(skip_serializing_if = "Option::is_none")]
51    pub port: Option<String>,
52    #[serde(skip_serializing_if = "Option::is_none")]
53    pub user_agent: Option<String>,
54}
55
56impl WatchSource {
57    /// Return whether the server supplied any source metadata.
58    pub fn is_empty(&self) -> bool {
59        self.event_source.is_none()
60            && self.region.is_none()
61            && self.principal_id.is_none()
62            && self.host.is_none()
63            && self.port.is_none()
64            && self.user_agent.is_none()
65    }
66}
67
68/// A version-operation-independent object notification record.
69#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
70pub struct WatchEvent {
71    pub event_id: Option<String>,
72    pub event_name: String,
73    pub bucket: String,
74    pub key: String,
75    pub version_id: Option<String>,
76    pub delete_marker: bool,
77    pub size_bytes: Option<i64>,
78    pub etag: Option<String>,
79    pub event_time: Timestamp,
80    #[serde(skip_serializing_if = "Option::is_none")]
81    pub source: Option<WatchSource>,
82}
83
84/// Backend boundary for opening RustFS-compatible live notification streams.
85#[async_trait]
86pub trait WatchApi: Send + Sync {
87    /// Open a stream using the supplied scope and server-side filters.
88    async fn watch(&self, request: &WatchRequest) -> Result<WatchStream>;
89}
90
91#[cfg(test)]
92mod tests {
93    use super::*;
94
95    #[test]
96    fn source_metadata_detects_empty_and_populated_values() {
97        assert!(WatchSource::default().is_empty());
98
99        let source = WatchSource {
100            host: Some("node-1".to_string()),
101            ..WatchSource::default()
102        };
103        assert!(!source.is_empty());
104    }
105
106    #[test]
107    fn watch_event_serialization_is_independent_of_object_version_operations() {
108        let event = WatchEvent {
109            event_id: Some("sequence-1".to_string()),
110            event_name: "s3:ObjectRemoved:DeleteMarkerCreated".to_string(),
111            bucket: "photos".to_string(),
112            key: "old.jpg".to_string(),
113            version_id: Some("v2".to_string()),
114            delete_marker: true,
115            size_bytes: None,
116            etag: None,
117            event_time: "2026-07-21T04:01:00Z"
118                .parse()
119                .expect("timestamp should parse"),
120            source: None,
121        };
122
123        let value = serde_json::to_value(event).expect("event should serialize");
124        assert_eq!(value["delete_marker"], true);
125        assert_eq!(value["version_id"], "v2");
126        assert!(value.get("is_latest").is_none());
127    }
128}
rc_core/watch.rs

rc_core/
watch.rs
