re_log_channel/
lib.rs

1//! An in-memory channel of Rerun data messages
2
3use std::sync::Arc;
4
5pub use crossbeam::channel::{RecvError, RecvTimeoutError, SendError, TryRecvError};
6use parking_lot::RwLock;
7pub use re_quota_channel::sync::TrySendError;
8use re_uri::RedapUri;
9
10mod data_source_message;
11mod receiver;
12mod receiver_set;
13mod sender;
14
15pub use self::data_source_message::{
16    BlueprintTarget, DataSourceMessage, DataSourceUiCommand, DefaultBlueprintRegistration,
17    InspectError, SaveScreenshotError,
18};
19pub use self::receiver::LogReceiver;
20pub use self::receiver_set::LogReceiverSet;
21pub use self::sender::LogSender;
22
23// --- Source ---
24
25/// Controls how a newly loaded recording is treated by the viewer.
26#[derive(
27    Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Deserialize, serde::Serialize,
28)]
29pub enum RecordingOpenBehavior {
30    /// Load without affecting the recording panel.
31    ///
32    /// Used for preview views.
33    Background,
34
35    /// Mark as opened in the recording panel, but don't navigate to it.
36    Open,
37
38    /// Mark as opened and make it the active recording.
39    OpenAndSelect,
40}
41
42/// An error that can occur when flushing.
43#[derive(Debug, thiserror::Error)]
44pub enum FlushError {
45    #[error("Received closed before flushing completed")]
46    Closed,
47
48    #[error("Flush timed out - not all messages were sent.")]
49    Timeout,
50}
51
52/// Identifies in what context this smart channel was created,
53/// and what is holding the [`LogSender`].
54#[derive(
55    Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Deserialize, serde::Serialize,
56)]
57pub enum LogSource {
58    /// The sender is a background thread reading data from a file on disk
59    /// (could be `.rrd` files, or `.glb`, `.png`, …).
60    File { path: std::path::PathBuf },
61
62    /// The sender is a background thread fetching data from an HTTP file server.
63    #[serde(alias = "RrdHttpStream")]
64    HttpStream {
65        /// Should include `http(s)://` prefix.
66        url: String,
67    },
68
69    /// The channel was created in the context of loading an `.rrd` file from a `postMessage`
70    /// javascript event.
71    ///
72    /// Only applicable to web browser iframes.
73    /// Used for the inline web viewer in a notebook.
74    RrdWebEvent,
75
76    /// The channel was created in the context of a javascript client submitting an RRD directly as bytes.
77    JsChannel {
78        /// The name of the channel reported by the javascript client.
79        channel_name: String,
80    },
81
82    /// The sender is a Rerun SDK running from another thread in the same process.
83    Sdk,
84
85    /// The data is streaming in from standard input.
86    Stdin,
87
88    /// The data is streaming in directly from a catalog server,
89    /// over `rerun://` gRPC interface.
90    RedapGrpcStream {
91        uri: re_uri::DatasetUri,
92
93        open_behavior: RecordingOpenBehavior,
94    },
95
96    /// The data is streaming in via a message proxy.
97    MessageProxy(re_uri::ProxyUri),
98}
99
100impl std::fmt::Display for LogSource {
101    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
102        match self {
103            Self::File { path } => write!(f, "file://{}", path.to_string_lossy()),
104            Self::HttpStream { url } => url_display_name(url).fmt(f),
105            Self::MessageProxy(uri) => uri.fmt(f),
106            Self::RedapGrpcStream { uri, .. } => uri.fmt(f),
107            Self::RrdWebEvent => "Web event listener".fmt(f),
108            Self::JsChannel { channel_name } => write!(f, "Javascript channel: {channel_name}"),
109            Self::Sdk => "SDK".fmt(f),
110            Self::Stdin => "stdin".fmt(f),
111        }
112    }
113}
114
115impl LogSource {
116    pub fn is_redap(&self) -> bool {
117        matches!(self, Self::RedapGrpcStream { .. })
118    }
119
120    pub fn is_network(&self) -> bool {
121        match self {
122            Self::File { .. } | Self::Sdk | Self::RrdWebEvent | Self::Stdin => false,
123            Self::HttpStream { .. }
124            | Self::JsChannel { .. }
125            | Self::RedapGrpcStream { .. }
126            | Self::MessageProxy { .. } => true,
127        }
128    }
129
130    pub fn open_behavior(&self) -> RecordingOpenBehavior {
131        match self {
132            Self::File { .. }
133            | Self::Sdk
134            | Self::RrdWebEvent
135            | Self::Stdin
136            | Self::HttpStream { .. }
137            | Self::JsChannel { .. }
138            | Self::MessageProxy { .. } => RecordingOpenBehavior::OpenAndSelect,
139
140            Self::RedapGrpcStream { open_behavior, .. } => *open_behavior,
141        }
142    }
143
144    pub fn redap_uri(&self) -> Option<RedapUri> {
145        match self {
146            Self::RedapGrpcStream { uri, .. } => Some(RedapUri::Dataset(uri.clone())),
147            Self::MessageProxy(uri) => Some(RedapUri::Proxy(uri.clone())),
148
149            Self::File { .. }
150            | Self::Sdk
151            | Self::RrdWebEvent
152            | Self::Stdin
153            | Self::HttpStream { .. }
154            | Self::JsChannel { .. } => None,
155        }
156    }
157
158    /// Same as [`Self::redap_uri`], but strips any fragment from the uri.
159    pub fn stripped_redap_uri(&self) -> Option<RedapUri> {
160        self.redap_uri().map(|uri| match uri {
161            RedapUri::Catalog(_)
162            | RedapUri::Entry(_)
163            | RedapUri::Folder(_)
164            | RedapUri::Proxy(_) => uri,
165            RedapUri::Dataset(uri) => RedapUri::Dataset(uri.without_fragment()),
166        })
167    }
168
169    /// Loading text for sources that load data from a specific source (e.g. a file or a URL).
170    ///
171    /// Returns `None` for any source that receives data dynamically through SDK calls or similar.
172    /// For a status string that applies to all sources, see [`Self::status_string`].
173    pub fn loading_name(&self) -> Option<String> {
174        match self {
175            // We only show things we know are very-soon-to-be recordings:
176            Self::File { path } => Some(path.to_string_lossy().into_owned()),
177            Self::HttpStream { url } => Some(url_display_name(url)),
178            Self::RedapGrpcStream { uri, .. } => uri
179                .segment_id
180                .as_ref()
181                .map(|segment_id| segment_id.as_str().to_owned()),
182
183            Self::RrdWebEvent
184            | Self::JsChannel { .. }
185            | Self::MessageProxy { .. }
186            | Self::Sdk
187            | Self::Stdin => {
188                // For all of these sources we're not actively loading data, but rather waiting for data to be sent.
189                // These show up in the top panel - see `top_panel.rs`.
190                None
191            }
192        }
193    }
194
195    /// Status string describing waiting or loading status for a source.
196    pub fn status_string(&self) -> String {
197        match self {
198            Self::File { path } => {
199                format!("Loading {}…", path.display())
200            }
201            Self::Stdin => "Loading stdin…".to_owned(),
202            Self::HttpStream { url } => {
203                format!("Waiting for data on {}…", url_display_name(url))
204            }
205            Self::MessageProxy(uri) => {
206                format!("Waiting for data on {uri}…")
207            }
208            Self::RedapGrpcStream { uri, .. } => {
209                format!("Waiting for data on {}…", uri.clone().without_fragment())
210            }
211            Self::RrdWebEvent | Self::JsChannel { .. } => "Waiting for logging data…".to_owned(),
212            Self::Sdk => "Waiting for logging data from SDK".to_owned(),
213        }
214    }
215
216    /// Compares two channel sources but ignores any URI fragments and other selection/state only guides
217    /// that don't affect what data is loaded.
218    pub fn is_same_ignoring_uri_fragments(&self, other: &Self) -> bool {
219        match (self, other) {
220            (Self::RedapGrpcStream { uri: uri1, .. }, Self::RedapGrpcStream { uri: uri2, .. }) => {
221                uri1.clone().without_fragment() == uri2.clone().without_fragment()
222            }
223            (Self::HttpStream { url: url1 }, Self::HttpStream { url: url2 }) => url1 == url2,
224            _ => self == other,
225        }
226    }
227}
228
229/// A human-readable name for a source URL, safe to render as a label or log line.
230///
231/// `data:` URLs embed their payload inline and can be many megabytes long.
232pub fn url_display_name(url: &str) -> String {
233    // The part of a `data:` URL before the first comma is the media type
234    // (e.g. `data:application/octet-stream;base64`); the rest is the payload.
235    if url.starts_with("data:")
236        && let Some(comma) = url.find(',')
237    {
238        return format!("{}…", &url[..=comma]);
239    }
240
241    url.to_owned()
242}
243
244// -------------------------------------------------------------------------------------
245
246/// Shared by all receivers and senders of a channel
247#[derive(Default)]
248pub(crate) struct Channel {
249    /// The sender should call this every time a message is sent.
250    ///
251    /// This can be used to wake up the receiver thread.
252    waker: RwLock<Option<Box<dyn Fn() + Send + Sync + 'static>>>,
253}
254
255/// Create a new communication channel for [`DataSourceMessage`].
256pub fn log_channel(source: LogSource) -> (LogSender, LogReceiver) {
257    let max_bytes_on_wire = 128 * 1024 * 1024; // TODO(emilk): make configurable
258
259    let source = Arc::new(source);
260    let channel = Arc::new(Channel::default());
261    let (tx, rx) = re_quota_channel::channel(format!("log_channel({source})"), max_bytes_on_wire);
262    let sender = LogSender::new(tx, source.clone(), channel.clone());
263    let receiver = LogReceiver::new(rx, channel, source);
264    (sender, receiver)
265}
266
267// ---
268
269/// The payload of a [`SmartMessage`].
270///
271/// Either data or an end-of-stream marker.
272#[derive(re_byte_size::SizeBytes)]
273pub enum SmartMessagePayload {
274    /// A message sent down the channel.
275    Msg(DataSourceMessage),
276
277    /// When received, flush anything already received and then call the given callback.
278    Flush {
279        #[size_bytes(ignore)]
280        on_flush_done: Box<dyn FnOnce() + Send>,
281    },
282
283    /// The [`LogSender`] has quit.
284    ///
285    /// `None` indicates the sender left gracefully, an error indicates otherwise.
286    Quit(#[size_bytes(ignore)] Option<Box<dyn std::error::Error + Send>>),
287}
288
289impl std::fmt::Debug for SmartMessagePayload {
290    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
291        match self {
292            Self::Msg(_) => f.write_str("Msg(_)"),
293            Self::Flush { .. } => f.write_str("Flush"),
294            Self::Quit(_) => f.write_str("Quit"),
295        }
296    }
297}
298
299#[derive(Debug, re_byte_size::SizeBytes)]
300pub struct SmartMessage {
301    #[size_bytes(ignore)]
302    pub source: Arc<LogSource>,
303    pub payload: SmartMessagePayload,
304}
305
306impl SmartMessage {
307    pub fn data(&self) -> Option<&DataSourceMessage> {
308        match &self.payload {
309            SmartMessagePayload::Msg(msg) => Some(msg),
310            SmartMessagePayload::Flush { .. } | SmartMessagePayload::Quit(_) => None,
311        }
312    }
313
314    pub fn into_data(self) -> Option<DataSourceMessage> {
315        match self.payload {
316            SmartMessagePayload::Msg(msg) => Some(msg),
317            SmartMessagePayload::Flush { .. } | SmartMessagePayload::Quit(_) => None,
318        }
319    }
320}
321
322#[cfg(test)]
323mod tests {
324    use super::url_display_name;
325
326    #[test]
327    fn url_display_name_keeps_short_urls() {
328        let url = "https://example.com/data.rrd";
329        assert_eq!(url_display_name(url), url);
330    }
331
332    #[test]
333    fn url_display_name_keeps_long_real_urls() {
334        // Presigned links and redap URIs are legitimately long — render them in full.
335        let url = format!("https://example.com/data.rrd?token={}", "x".repeat(1000));
336        assert_eq!(url_display_name(&url), url);
337    }
338
339    #[test]
340    fn url_display_name_truncates_long_data_url() {
341        // A multi-megabyte `data:` URL must not be rendered verbatim (it OOMs text layout).
342        let payload = "A".repeat(5_000_000);
343        let url = format!("data:application/octet-stream;base64,{payload}");
344
345        let name = url_display_name(&url);
346
347        assert_eq!(name, "data:application/octet-stream;base64,…");
348        assert!(name.len() < 100);
349    }
350}
re_log_channel/lib.rs

re_log_channel/
lib.rs
