Standard input/output support 1: stream RRD data from stdin

Cargo.toml

@@ -101,6 +101,7 @@ array-init = "2.1"
 arrow2 = "0.17"
 arrow2_convert = "0.5.0"
 async-executor = "1.0"
+atty = "0.2"
 backtrace = "0.3"
 bincode = "1.3"
 bitflags = { version = "2.4", features = ["bytemuck"] }

crates/re_data_source/src/data_source.rs

@@ -22,6 +22,10 @@ pub enum DataSource {
 
     /// A remote Rerun server.
     WebSocketAddr(String),
+
+    // RRD data streaming in from standard input.
+    #[cfg(not(target_arch = "wasm32"))]
+    Stdin,
 }
 
 impl DataSource {
@@ -65,6 +69,16 @@ impl DataSource {
             }
         }
 
+        // Reading from standard input in non-TTY environments (e.g. GitHub Actions, but I'm sure we can
+        // come up with more convoluted than that…) can lead to many unexpected,
+        // platform-specific problems that aren't even necessarily consistent across runs.
+        //
+        // In order to avoid having to swallow errors based on unreliable heuristics (or inversely:
+        // throwing errors when we shouldn't), we just make reading from standard input explicit.
+        if uri == "-" {
+            return DataSource::Stdin;
+        }
+
         let path = std::path::Path::new(&uri).to_path_buf();
 
         if uri.starts_with("file://") || path.exists() {
@@ -147,6 +161,22 @@ impl DataSource {
             DataSource::WebSocketAddr(rerun_server_ws_url) => {
                 crate::web_sockets::connect_to_ws_url(&rerun_server_ws_url, on_msg)
             }
+
+            #[cfg(not(target_arch = "wasm32"))]
+            DataSource::Stdin => {
+                let (tx, rx) = re_smart_channel::smart_channel(
+                    SmartMessageSource::Stdin,
+                    SmartChannelSource::Stdin,
+                );
+
+                crate::load_stdin::load_stdin(tx).with_context(|| "stdin".to_owned())?;
+
+                if let Some(on_msg) = on_msg {
+                    on_msg();
+                }
+
+                Ok(rx)
+            }
         }
     }
 }

crates/re_data_source/src/lib.rs

@@ -16,6 +16,9 @@ mod load_file;
 mod load_file_contents;
 mod web_sockets;
 
+#[cfg(not(target_arch = "wasm32"))]
+mod load_stdin;
+
 #[cfg(not(target_arch = "wasm32"))]
 mod load_file_path;
 

crates/re_data_source/src/load_stdin.rs

@@ -0,0 +1,33 @@
+use re_log_types::LogMsg;
+use re_smart_channel::Sender;
+
+/// Asynchronously loads RRD data streaming in from standard input.
+///
+/// This fails synchronously iff the standard input stream could not be opened, otherwise errors
+/// are handlded asynchronously (as in: they're logged).
+pub fn load_stdin(tx: Sender<LogMsg>) -> anyhow::Result<()> {
+    let version_policy = re_log_encoding::decoder::VersionPolicy::Warn;
+
+    let decoder = re_log_encoding::decoder::Decoder::new(version_policy, std::io::stdin())?;
+
+    rayon::spawn(move || {
+        re_tracing::profile_scope!("stdin");
+
+        for msg in decoder {
+            let msg = match msg {
+                Ok(msg) => msg,
+                Err(err) => {
+                    re_log::warn_once!("Failed to decode message in stdin: {err}");
+                    continue;
+                }
+            };
+            if tx.send(msg).is_err() {
+                break; // The other end has decided to hang up, not our problem.
+            }
+        }
+
+        tx.quit(None).ok(); // The other end has decided to hang up, not our problem.
+    });
+
+    Ok(())
+}

crates/re_smart_channel/src/lib.rs

@@ -50,12 +50,15 @@ pub enum SmartChannelSource {
     ///
     /// We are a TCP server listening on this port.
     TcpServer { port: u16 },
+
+    /// The channel was created in the context of streaming in RRD data from standard input.
+    Stdin,
 }
 
 impl SmartChannelSource {
     pub fn is_network(&self) -> bool {
         match self {
-            Self::File(_) | Self::Sdk | Self::RrdWebEventListener => false,
+            Self::File(_) | Self::Sdk | Self::RrdWebEventListener | Self::Stdin => false,
             Self::RrdHttpStream { .. } | Self::WsClient { .. } | Self::TcpServer { .. } => true,
         }
     }
@@ -103,6 +106,9 @@ pub enum SmartMessageSource {
         // reason.
         addr: Option<std::net::SocketAddr>,
     },
+
+    /// The data is streaming in from standard input.
+    Stdin,
 }
 
 impl std::fmt::Display for SmartMessageSource {
@@ -118,6 +124,7 @@ impl std::fmt::Display for SmartMessageSource {
                 "tcp://{}",
                 addr.map_or_else(|| "(unknown ip)".to_owned(), |addr| addr.to_string())
             ),
+            SmartMessageSource::Stdin => "stdin".into(),
         })
     }
 }

crates/re_viewer/src/app.rs

@@ -974,7 +974,9 @@ impl App {
         for source in sources {
             match &*source {
                 // No need for a welcome screen - data is coming soon!
-                SmartChannelSource::File(_) | SmartChannelSource::RrdHttpStream { .. } => {
+                SmartChannelSource::File(_)
+                | SmartChannelSource::RrdHttpStream { .. }
+                | SmartChannelSource::Stdin => {
                     return false;
                 }
 

crates/re_viewer/src/app_state.rs

@@ -327,7 +327,8 @@ fn recording_config_entry<'cfgs>(
                 // Live data - follow it!
                 re_smart_channel::SmartChannelSource::Sdk
                 | re_smart_channel::SmartChannelSource::WsClient { .. }
-                | re_smart_channel::SmartChannelSource::TcpServer { .. } => PlayState::Following,
+                | re_smart_channel::SmartChannelSource::TcpServer { .. }
+                | re_smart_channel::SmartChannelSource::Stdin => PlayState::Following,
             }
         } else {
             PlayState::Following // No known source 🤷‍♂️

crates/re_viewer/src/ui/recordings_panel.rs

@@ -68,7 +68,8 @@ fn loading_receivers_ui(
             SmartChannelSource::RrdWebEventListener
             | SmartChannelSource::Sdk
             | SmartChannelSource::WsClient { .. }
-            | SmartChannelSource::TcpServer { .. } => {
+            | SmartChannelSource::TcpServer { .. }
+            | SmartChannelSource::Stdin => {
                 // These show up in the top panel - see `top_panel.rs`.
                 continue;
             }
@@ -294,6 +295,7 @@ fn data_source_string(data_source: &re_smart_channel::SmartChannelSource) -> Str
         SmartChannelSource::Sdk => "SDK".to_owned(),
         SmartChannelSource::WsClient { ws_server_url } => ws_server_url.clone(),
         SmartChannelSource::TcpServer { port } => format!("TCP Server, port {port}"),
+        SmartChannelSource::Stdin => "Standard Input".to_owned(),
     }
 }
 

crates/re_viewer/src/ui/top_panel.rs

@@ -99,7 +99,9 @@ fn connection_status_ui(ui: &mut egui::Ui, rx: &ReceiveSet<re_log_types::LogMsg>
         .into_iter()
         .filter(|source| {
             match source.as_ref() {
-                SmartChannelSource::File(_) | SmartChannelSource::RrdHttpStream { .. } => {
+                SmartChannelSource::File(_)
+                | SmartChannelSource::RrdHttpStream { .. }
+                | SmartChannelSource::Stdin => {
                     false // These show up in the recordings panel as a "Loading…" in `recordings_panel.rs`
                 }
 
@@ -134,6 +136,7 @@ fn connection_status_ui(ui: &mut egui::Ui, rx: &ReceiveSet<re_log_types::LogMsg>
 
         let tooltip = match source {
             SmartChannelSource::File(_)
+            | SmartChannelSource::Stdin
             | SmartChannelSource::RrdHttpStream { .. }
             | SmartChannelSource::RrdWebEventListener
             | SmartChannelSource::Sdk
@@ -156,6 +159,7 @@ fn connection_status_ui(ui: &mut egui::Ui, rx: &ReceiveSet<re_log_types::LogMsg>
             re_smart_channel::SmartChannelSource::File(path) => {
                 format!("Loading {}…", path.display())
             }
+            re_smart_channel::SmartChannelSource::Stdin => "Loading stdin…".to_owned(),
             re_smart_channel::SmartChannelSource::RrdHttpStream { url } => {
                 format!("Loading {url}…")
             }

crates/re_viewer/src/viewer_analytics.rs

@@ -233,6 +233,7 @@ impl ViewerAnalytics {
                 re_smart_channel::SmartChannelSource::Sdk => "sdk", // show()
                 re_smart_channel::SmartChannelSource::WsClient { .. } => "ws_client", // spawn()
                 re_smart_channel::SmartChannelSource::TcpServer { .. } => "tcp_server", // connect()
+                re_smart_channel::SmartChannelSource::Stdin => "stdin",
             };
             self.register("data_source", data_source);
         }

crates/rerun/src/lib.rs

@@ -142,6 +142,7 @@ pub mod external {
     pub use anyhow;
 
     pub use re_build_info;
+    pub use re_data_source;
     pub use re_data_store;
     pub use re_data_store::external::*;
     pub use re_format;