ReDap Catalog as data source

Cargo.lock

@@ -5789,6 +5789,7 @@ dependencies = [
  "tokio-stream",
  "tonic",
  "tonic-web-wasm-client",
+ "url",
  "wasm-bindgen-futures",
 ]
 

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 _;
 
@@ -32,7 +31,8 @@ pub enum DataSource {
     #[cfg(not(target_arch = "wasm32"))]
     Stdin,
 
-    /// A file on a Rerun Data Platform server, over `rerun://` gRPC interface.
+    /// A file or a metadata catalog on a Rerun Data Platform server,
+    /// over `rerun://` gRPC interface.
     #[cfg(feature = "grpc")]
     RerunGrpcUrl { url: String },
 }
@@ -247,7 +247,7 @@ impl DataSource {
 
             #[cfg(feature = "grpc")]
             Self::RerunGrpcUrl { url } => {
-                re_grpc_client::stream_recording(url, on_msg).map_err(|err| err.into())
+                re_grpc_client::stream_from_redap(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,162 @@
+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 {
+#[error("URL {url:?} should follow rerun://addr:port/recording/12345 for recording or rerun://addr:port/catalog for catalog")]
+pub struct InvalidRedapAddress {
     url: String,
     msg: String,
 }
 
-/// Parsed `rerun://addr:port/recording/12345`
-pub struct Address {
-    pub addr_port: String,
-    pub recording_id: String,
+/// Parsed from `rerun://addr:port/recording/12345` or `rerun://addr:port/catalog`
+#[derive(PartialEq, Eq, Debug)]
+pub enum RedapAddress {
+    Recording {
+        redap_endpoint: Url,
+        recording_id: String,
+    },
+    Catalog {
+        redap_endpoint: Url,
+    },
 }
 
-impl std::fmt::Display for Address {
+impl std::fmt::Display for RedapAddress {
+    #[allow(clippy::unwrap_used)] // host and port have already been verified during conversion
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "rerun://{}/recording/{}",
-            self.addr_port, self.recording_id
-        )
+        match self {
+            Self::Recording {
+                redap_endpoint,
+                recording_id,
+            } => write!(
+                f,
+                "rerun://{}:{}/recording/{}",
+                redap_endpoint.host().unwrap(),
+                redap_endpoint.port().unwrap(),
+                recording_id
+            ),
+            Self::Catalog { redap_endpoint } => write!(
+                f,
+                "rerun://{}:{}/catalog",
+                redap_endpoint.host().unwrap(),
+                redap_endpoint.port().unwrap(),
+            ),
+        }
     }
 }
 
-impl std::str::FromStr for Address {
-    type Err = InvalidAddressError;
+impl TryFrom<&str> for RedapAddress {
+    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(value: &str) -> Result<Self, Self::Error> {
+        let url = Url::parse(value).map_err(|err| InvalidRedapAddress {
+            url: value.to_owned(),
+            msg: err.to_string(),
+        })?;
 
-        let parts = stripped_url.split('/').collect::<Vec<_>>();
-        if parts.len() < 3 {
-            return Err(InvalidAddressError {
-                url: url.to_owned(),
-                msg: "Too few slashes".to_owned(),
+        if url.scheme() != "rerun" {
+            return Err(InvalidRedapAddress {
+                url: url.to_string(),
+                msg: "Invalid scheme, expected 'rerun'".to_owned(),
             });
         }
-        if parts.len() > 3 {
-            return Err(InvalidAddressError {
-                url: url.to_owned(),
-                msg: "Too many slashes".to_owned(),
-            });
+
+        let host = url.host_str().ok_or_else(|| InvalidRedapAddress {
+            url: url.to_string(),
+            msg: "Missing host".to_owned(),
+        })?;
+
+        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.");
         }
 
-        if parts[1] != "recording" {
-            return Err(InvalidAddressError {
-                url: url.to_owned(),
-                msg: "Not a recording".to_owned(),
-            });
+        let port = url.port().ok_or_else(|| InvalidRedapAddress {
+            url: url.to_string(),
+            msg: "Missing port".to_owned(),
+        })?;
+
+        #[allow(clippy::unwrap_used)]
+        let redap_endpoint = Url::parse(&format!("http://{host}:{port}")).unwrap();
+
+        // we got the ReDap endpoint, now figure out from the URL path if it's a recording or catalog
+        if url.path().ends_with("/catalog") {
+            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.to_string(),
+                    msg: "Invalid path, expected '/catalog'".to_owned(),
+                });
+            }
+
+            Ok(Self::Catalog { redap_endpoint })
+        } else {
+            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.to_string(),
+                    msg: "Invalid path, expected '/recording/{id}'".to_owned(),
+                });
+            }
+
+            Ok(Self::Recording {
+                redap_endpoint,
+                recording_id: path_segments[1].to_owned(),
+            })
         }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use super::RedapAddress;
+    use url::Url;
+
+    #[test]
+    fn test_recording_url_to_address() {
+        let url = "rerun://0.0.0.0:1234/recording/12345";
+        let address: RedapAddress = url.try_into().unwrap();
+        assert_eq!(
+            address,
+            RedapAddress::Recording {
+                redap_endpoint: Url::parse("http://0.0.0.0:1234").unwrap(),
+                recording_id: "12345".to_owned()
+            }
+        );
+    }
+
+    #[test]
+    fn test_catalog_url_to_address() {
+        let url = "rerun://127.0.0.1:50051/catalog";
+        let address: RedapAddress = url.try_into().unwrap();
+        assert_eq!(
+            address,
+            RedapAddress::Catalog {
+                redap_endpoint: Url::parse("http://127.0.0.1:50051").unwrap(),
+            }
+        );
+    }
+
+    #[test]
+    fn test_invalid_url() {
+        let url = "http://wrong-scheme:1234/recording/12345";
+        let address: Result<RedapAddress, _> = url.try_into();
+
+        assert!(matches!(
+            address.unwrap_err(),
+            super::InvalidRedapAddress { .. }
+        ));
+    }
 
-        let addr_port = parts[0].to_owned();
-        let recording_id = parts[2].to_owned();
+    #[test]
+    fn test_invalid_path() {
+        let url = "rerun://0.0.0.0:51234/redap/recordings/12345";
+        let address: Result<RedapAddress, _> = url.try_into();
 
-        Ok(Self {
-            addr_port,
-            recording_id,
-        })
+        assert!(matches!(
+            address.unwrap_err(),
+            super::InvalidRedapAddress { .. }
+        ));
     }
 }