ReDap Catalog as data source

Cargo.lock

@@ -5636,6 +5636,7 @@ dependencies = [
  "re_smart_channel",
  "re_tracing",
  "re_ws_comms",
+ "url",
 ]
 
 [[package]]
@@ -5789,6 +5790,7 @@ dependencies = [
  "tokio-stream",
  "tonic",
  "tonic-web-wasm-client",
+ "url",
  "wasm-bindgen-futures",
 ]
 

crates/store/re_data_source/Cargo.toml

@@ -41,6 +41,7 @@ re_ws_comms = { workspace = true, features = ["client"] }
 anyhow.workspace = true
 itertools.workspace = true
 rayon.workspace = true
+url.workspace = true
 
 # Optional dependencies:
 re_grpc_client = { workspace = true, optional = true }

crates/store/re_data_source/src/data_source.rs

@@ -1,8 +1,7 @@
+use crate::FileContents;
 use re_log_types::LogMsg;
 use re_smart_channel::{Receiver, SmartChannelSource, SmartMessageSource};
 
-use crate::FileContents;
-
 #[cfg(not(target_arch = "wasm32"))]
 use anyhow::Context as _;
 
@@ -35,6 +34,10 @@ pub enum DataSource {
     /// A file on a Rerun Data Platform server, over `rerun://` gRPC interface.
     #[cfg(feature = "grpc")]
     RerunGrpcUrl { url: String },
+
+    /// A catalog containing metadata information about recordings stored on Rerun Data Platform.
+    #[cfg(feature = "grpc")]
+    RerunGrpcCatalog { url: String },
 }
 
 impl DataSource {
@@ -92,6 +95,9 @@ impl DataSource {
 
         #[cfg(feature = "grpc")]
         if uri.starts_with("rerun://") {
+            if uri.ends_with("/catalog") {
+                return Self::RerunGrpcCatalog { url: uri };
+            }
             return Self::RerunGrpcUrl { url: uri };
         }
 
@@ -137,7 +143,7 @@ impl DataSource {
             #[cfg(not(target_arch = "wasm32"))]
             Self::Stdin => None,
             #[cfg(feature = "grpc")]
-            Self::RerunGrpcUrl { .. } => None, // TODO(jleibs): This needs to come from the server.
+            Self::RerunGrpcUrl { .. } | Self::RerunGrpcCatalog { .. } => None, // TODO(jleibs): This needs to come from the server.
         }
     }
 
@@ -247,8 +253,17 @@ impl DataSource {
 
             #[cfg(feature = "grpc")]
             Self::RerunGrpcUrl { url } => {
+                let url =
+                    url::Url::parse(&url).with_context(|| format!("Invalid gRPC URL: {url}"))?;
                 re_grpc_client::stream_recording(url, on_msg).map_err(|err| err.into())
             }
+
+            #[cfg(feature = "grpc")]
+            Self::RerunGrpcCatalog { url } => {
+                let url =
+                    url::Url::parse(&url).with_context(|| format!("Invalid gRPC URL: {url}"))?;
+                re_grpc_client::stream_catalog(url, on_msg).map_err(|err| err.into())
+            }
         }
     }
 }

crates/store/re_grpc_client/Cargo.toml

@@ -29,6 +29,7 @@ re_smart_channel.workspace = true
 
 thiserror.workspace = true
 tokio-stream.workspace = true
+url.workspace = true
 
 # Native dependencies:
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]

crates/store/re_grpc_client/src/address.rs

@@ -1,65 +1,111 @@
+use url::Url;
+
 /// The given url is not a valid Rerun storage node URL.
 #[derive(thiserror::Error, Debug)]
-#[error("URL {url:?} should follow rerun://addr:port/recording/12345")]
-pub struct InvalidAddressError {
-    url: String,
+#[error("URL {url:?} should follow rerun://addr:port/recording/12345 for recording or rerun://addr:port/catalog for catalog")]
+pub struct InvalidRedapAddress {
+    url: Url,
     msg: String,
 }
 
 /// Parsed `rerun://addr:port/recording/12345`
-pub struct Address {
-    pub addr_port: String,
+pub struct RecordingAddress {
+    pub redap_endpoint: Url,
     pub recording_id: String,
 }
 
-impl std::fmt::Display for Address {
+impl std::fmt::Display for RecordingAddress {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "rerun://{}/recording/{}",
-            self.addr_port, self.recording_id
+            "RecordingAddress {{ redap_endpoint: {}, recording_id: {} }}",
+            self.redap_endpoint, self.recording_id
         )
     }
 }
 
-impl std::str::FromStr for Address {
-    type Err = InvalidAddressError;
+impl TryFrom<Url> for RecordingAddress {
+    type Error = InvalidRedapAddress;
 
-    fn from_str(url: &str) -> Result<Self, Self::Err> {
-        let Some(stripped_url) = url.strip_prefix("rerun://") else {
-            return Err(InvalidAddressError {
-                url: url.to_owned(),
-                msg: "Missing rerun://".to_owned(),
+    fn try_from(url: Url) -> Result<Self, Self::Error> {
+        if url.scheme() != "rerun" {
+            return Err(InvalidRedapAddress {
+                url: url.clone(),
+                msg: "Invalid scheme, expected 'rerun'".to_owned(),
             });
-        };
+        }
 
-        let parts = stripped_url.split('/').collect::<Vec<_>>();
-        if parts.len() < 3 {
-            return Err(InvalidAddressError {
-                url: url.to_owned(),
-                msg: "Too few slashes".to_owned(),
+        let redap_endpoint = get_redap_endpoint(&url)?;
+
+        let path_segments: Vec<&str> = url.path_segments().map(|s| s.collect()).unwrap_or_default();
+        if path_segments.len() != 2 || path_segments[0] != "recording" {
+            return Err(InvalidRedapAddress {
+                url: url.clone(),
+                msg: "Invalid path, expected '/recording/{id}'".to_owned(),
             });
         }
-        if parts.len() > 3 {
-            return Err(InvalidAddressError {
-                url: url.to_owned(),
-                msg: "Too many slashes".to_owned(),
+
+        Ok(Self {
+            redap_endpoint,
+            recording_id: path_segments[1].to_owned(),
+        })
+    }
+}
+
+/// Parsed `rerun://addr:port/catalog`
+pub struct CatalogAddress {
+    pub redap_endpoint: Url,
+}
+
+impl std::fmt::Display for CatalogAddress {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "ReDap endpoint {}", self.redap_endpoint)
+    }
+}
+
+impl TryFrom<Url> for CatalogAddress {
+    type Error = InvalidRedapAddress;
+
+    fn try_from(url: Url) -> Result<Self, Self::Error> {
+        if url.scheme() != "rerun" {
+            return Err(InvalidRedapAddress {
+                url: url.clone(),
+                msg: "Invalid scheme, expected 'rerun'".to_owned(),
             });
         }
 
-        if parts[1] != "recording" {
-            return Err(InvalidAddressError {
-                url: url.to_owned(),
-                msg: "Not a recording".to_owned(),
+        let redap_endpoint = get_redap_endpoint(&url)?;
+
+        let path_segments: Vec<&str> = url.path_segments().map(|s| s.collect()).unwrap_or_default();
+        if path_segments.len() != 1 || path_segments[0] != "catalog" {
+            return Err(InvalidRedapAddress {
+                url: url.clone(),
+                msg: "Invalid path, expected '/catalog'".to_owned(),
             });
         }
 
-        let addr_port = parts[0].to_owned();
-        let recording_id = parts[2].to_owned();
+        Ok(Self { redap_endpoint })
+    }
+}
+
+/// Small helper to extract host and port from the Rerun Data Platform URL.
+fn get_redap_endpoint(url: &Url) -> Result<Url, InvalidRedapAddress> {
+    let host = url.host_str().ok_or_else(|| InvalidRedapAddress {
+        url: url.clone(),
+        msg: "Missing host".to_owned(),
+    })?;
 
-        Ok(Self {
-            addr_port,
-            recording_id,
-        })
+    if host == "0.0.0.0" {
+        re_log::warn!("Using 0.0.0.0 as Rerun Data Platform host will often fail. You might want to try using 127.0.0.0.");
     }
+
+    let port = url.port().ok_or_else(|| InvalidRedapAddress {
+        url: url.clone(),
+        msg: "Missing port".to_owned(),
+    })?;
+
+    #[allow(clippy::unwrap_used)]
+    let redap_endpoint = Url::parse(&format!("http://{host}:{port}")).unwrap();
+
+    Ok(redap_endpoint)
 }