First iteration of a dedicated UI and communication for the Redap server/catalog/collection browser (#9018)

### Related

* Closes: #9003

### What

Implement a whole new Redap server/catalog/collection UI.

---------

Co-authored-by: Andreas Reich <[email protected]>

Author: abey79

ARCHITECTURE.md

@@ -98,11 +98,11 @@ Of course, this will only take us so far. In the future we plan on caching queri
 Here is an overview of the crates included in the project:
 
 <picture>
-  <img src="https://static.rerun.io/crates/161078ff2b88caf8065e4bf8fb631a318329f501/full.png" alt="">
-  <source media="(max-width: 480px)" srcset="https://static.rerun.io/crates/161078ff2b88caf8065e4bf8fb631a318329f501/480w.png">
-  <source media="(max-width: 768px)" srcset="https://static.rerun.io/crates/161078ff2b88caf8065e4bf8fb631a318329f501/768w.png">
-  <source media="(max-width: 1024px)" srcset="https://static.rerun.io/crates/161078ff2b88caf8065e4bf8fb631a318329f501/1024w.png">
-  <source media="(max-width: 1200px)" srcset="https://static.rerun.io/crates/161078ff2b88caf8065e4bf8fb631a318329f501/1200w.png">
+  <img src="https://static.rerun.io/crates/4ebb81d4dc640dfb2634d37fec5c53e80fb25f51/full.png" alt="">
+  <source media="(max-width: 480px)" srcset="https://static.rerun.io/crates/4ebb81d4dc640dfb2634d37fec5c53e80fb25f51/480w.png">
+  <source media="(max-width: 768px)" srcset="https://static.rerun.io/crates/4ebb81d4dc640dfb2634d37fec5c53e80fb25f51/768w.png">
+  <source media="(max-width: 1024px)" srcset="https://static.rerun.io/crates/4ebb81d4dc640dfb2634d37fec5c53e80fb25f51/1024w.png">
+  <source media="(max-width: 1200px)" srcset="https://static.rerun.io/crates/4ebb81d4dc640dfb2634d37fec5c53e80fb25f51/1200w.png">
 </picture>
 
 <!-- !!! IMPORTANT!!!
@@ -134,8 +134,9 @@ Update instructions:
 ##### UI crates
 
 | Crate                 | Description                                                                                                |
-| --------------------- | ---------------------------------------------------------------------------------------------------------- |
+|-----------------------|------------------------------------------------------------------------------------------------------------|
 | re_blueprint_tree     | The UI for the blueprint tree in the left panel.                                                           |
+| re_catalog_hub        | The UI and communication to implement the in-viewer catalog browser                                        |
 | re_chunk_store_ui     | A chunk store browser UI.                                                                                  |
 | re_component_ui       | Provides UI editors for Rerun component data for registration with the Rerun Viewer component UI registry. |
 | re_selection_panel    | The UI for the selection panel.                                                                            |

Cargo.lock

@@ -5816,6 +5816,34 @@ dependencies = [
  "convert_case",
 ]
 
+[[package]]
+name = "re_catalog_hub"
+version = "0.23.0-alpha.1+dev"
+dependencies = [
+ "ahash",
+ "arrow",
+ "egui",
+ "egui_table",
+ "once_cell",
+ "parking_lot",
+ "re_arrow_util",
+ "re_grpc_client",
+ "re_log",
+ "re_log_encoding",
+ "re_log_types",
+ "re_protos",
+ "re_sorbet",
+ "re_types",
+ "re_types_core",
+ "re_ui",
+ "re_view_dataframe",
+ "re_viewer_context",
+ "thiserror 1.0.65",
+ "tokio-stream",
+ "tonic",
+ "tonic-web-wasm-client",
+]
+
 [[package]]
 name = "re_chunk"
 version = "0.23.0-alpha.1+dev"
@@ -6000,12 +6028,15 @@ dependencies = [
  "rayon",
  "re_build_tools",
  "re_data_loader",
+ "re_error",
  "re_grpc_client",
  "re_log",
  "re_log_encoding",
  "re_log_types",
  "re_smart_channel",
  "re_tracing",
+ "tokio",
+ "wasm-bindgen-futures",
 ]
 
 [[package]]
@@ -6169,7 +6200,6 @@ dependencies = [
  "re_protos",
  "re_smart_channel",
  "re_sorbet",
- "re_types",
  "thiserror 1.0.65",
  "tokio",
  "tokio-stream",
@@ -6889,6 +6919,7 @@ dependencies = [
  "re_log",
  "re_log_types",
  "re_renderer",
+ "re_sorbet",
  "re_tracing",
  "re_types",
  "re_types_core",
@@ -7101,6 +7132,7 @@ dependencies = [
  "re_build_info",
  "re_build_tools",
  "re_capabilities",
+ "re_catalog_hub",
  "re_chunk",
  "re_chunk_store",
  "re_chunk_store_ui",
@@ -7207,6 +7239,7 @@ dependencies = [
  "smallvec",
  "strum_macros",
  "thiserror 1.0.65",
+ "tokio",
  "uuid",
  "wasm-bindgen-futures",
  "web-sys",

Cargo.toml

@@ -91,6 +91,7 @@ re_video = { path = "crates/utils/re_video", version = "=0.23.0-alpha.1", defaul
 
 # crates/viewer:
 re_blueprint_tree = { path = "crates/viewer/re_blueprint_tree", version = "=0.23.0-alpha.1", default-features = false }
+re_catalog_hub = { path = "crates/viewer/re_catalog_hub", version = "=0.23.0-alpha.1", default-features = false }
 re_component_ui = { path = "crates/viewer/re_component_ui", version = "=0.23.0-alpha.1", default-features = false }
 re_context_menu = { path = "crates/viewer/re_context_menu", version = "=0.23.0-alpha.1", default-features = false }
 re_chunk_store_ui = { path = "crates/viewer/re_chunk_store_ui", version = "=0.23.0-alpha.1", default-features = false }

bacon.toml

@@ -86,3 +86,4 @@ c = "job:clippy"
 d = "job:doc-open"
 t = "job:test"
 r = "job:run"
+a = "job:wasm"

crates/store/re_data_source/Cargo.toml

@@ -28,6 +28,7 @@ grpc = ["re_grpc_client/redap"]
 
 [dependencies]
 re_data_loader.workspace = true
+re_error.workspace = true
 re_grpc_client.workspace = true
 re_log_encoding = { workspace = true, features = [
   "decoder",
@@ -42,7 +43,13 @@ anyhow.workspace = true
 itertools.workspace = true
 rayon.workspace = true
 
-# Optional dependencies:
+# native dependencies:
+[target.'cfg(not(target_arch = "wasm32"))'.dependencies]
+tokio.workspace = true
+
+# web dependencies:
+[target.'cfg(target_arch = "wasm32")'.dependencies]
+wasm-bindgen-futures.workspace = true
 
 [build-dependencies]
 re_build_tools.workspace = true

crates/store/re_data_source/src/data_source.rs

@@ -37,6 +37,14 @@ pub enum DataSource {
     MessageProxy { url: String },
 }
 
+// TODO(#9058): Temporary hack, see issue for how to fix this.
+pub enum StreamSource {
+    LogMessages(Receiver<LogMsg>),
+    CatalogData {
+        origin: re_grpc_client::redap::Origin,
+    },
+}
+
 impl DataSource {
     /// Tried to classify a URI into a [`DataSource`].
     ///
@@ -156,14 +164,15 @@ impl DataSource {
     pub fn stream(
         self,
         on_msg: Option<Box<dyn Fn() + Send + Sync>>,
-    ) -> anyhow::Result<Receiver<LogMsg>> {
+    ) -> anyhow::Result<StreamSource> {
         re_tracing::profile_function!();
+
         match self {
-            Self::RrdHttpUrl { url, follow } => Ok(
+            Self::RrdHttpUrl { url, follow } => Ok(StreamSource::LogMessages(
                 re_log_encoding::stream_rrd_from_http::stream_rrd_from_http_to_channel(
                     url, follow, on_msg,
                 ),
-            ),
+            )),
 
             #[cfg(not(target_arch = "wasm32"))]
             Self::FilePath(file_source, path) => {
@@ -190,7 +199,7 @@ impl DataSource {
                     on_msg();
                 }
 
-                Ok(rx)
+                Ok(StreamSource::LogMessages(rx))
             }
 
             // When loading a file on Web, or when using drag-n-drop.
@@ -224,7 +233,7 @@ impl DataSource {
                     on_msg();
                 }
 
-                Ok(rx)
+                Ok(StreamSource::LogMessages(rx))
             }
 
             #[cfg(not(target_arch = "wasm32"))]
@@ -240,17 +249,54 @@ impl DataSource {
                     on_msg();
                 }
 
-                Ok(rx)
+                Ok(StreamSource::LogMessages(rx))
             }
 
             #[cfg(feature = "grpc")]
             Self::RerunGrpcUrl { url } => {
-                re_grpc_client::redap::stream_from_redap(url, on_msg).map_err(|err| err.into())
+                use re_grpc_client::redap::RedapAddress;
+
+                re_log::debug!("Loading {url}…");
+
+                let address = url.as_str().try_into()?;
+
+                match address {
+                    RedapAddress::Recording {
+                        origin,
+                        recording_id,
+                    } => {
+                        let (tx, rx) = re_smart_channel::smart_channel(
+                            re_smart_channel::SmartMessageSource::RerunGrpcStream {
+                                url: url.clone(),
+                            },
+                            re_smart_channel::SmartChannelSource::RerunGrpcStream {
+                                url: url.clone(),
+                            },
+                        );
+                        spawn_future(async move {
+                            if let Err(err) = re_grpc_client::redap::stream_recording_async(
+                                tx,
+                                origin,
+                                recording_id,
+                                on_msg,
+                            )
+                            .await
+                            {
+                                re_log::warn!(
+                                    "Error while streaming {url}: {}",
+                                    re_error::format_ref(&err)
+                                );
+                            }
+                        });
+                        Ok(StreamSource::LogMessages(rx))
+                    }
+                    RedapAddress::Catalog { origin } => Ok(StreamSource::CatalogData { origin }),
+                }
             }
 
-            Self::MessageProxy { url } => {
-                re_grpc_client::message_proxy::stream(&url, on_msg).map_err(|err| err.into())
-            }
+            Self::MessageProxy { url } => re_grpc_client::message_proxy::stream(&url, on_msg)
+                .map_err(|err| err.into())
+                .map(StreamSource::LogMessages),
         }
     }
 }
@@ -319,3 +365,25 @@ fn test_data_source_from_uri() {
 
     assert!(!failed, "one or more test cases failed");
 }
+
+// TODO(ab, andreas): This should be replaced by the use of `AsyncRuntimeHandle`. However, this
+// requires:
+// - `AsyncRuntimeHandle` to be moved lower in the crate hierarchy to be available here (unsure
+//   where).
+// - Make sure that all callers of `DataSource::stream` have access to an `AsyncRuntimeHandle`
+//   (maybe it should be in `GlobalContext`?).
+#[cfg(target_arch = "wasm32")]
+fn spawn_future<F>(future: F)
+where
+    F: std::future::Future<Output = ()> + 'static,
+{
+    wasm_bindgen_futures::spawn_local(future);
+}
+
+#[cfg(not(target_arch = "wasm32"))]
+fn spawn_future<F>(future: F)
+where
+    F: std::future::Future<Output = ()> + 'static + Send,
+{
+    tokio::spawn(future);
+}

crates/store/re_data_source/src/lib.rs

@@ -11,7 +11,7 @@ mod data_source;
 #[cfg(not(target_arch = "wasm32"))]
 mod load_stdin;
 
-pub use self::data_source::DataSource;
+pub use self::data_source::{DataSource, StreamSource};
 
 // ----------------------------------------------------------------------------
 

crates/store/re_grpc_client/Cargo.toml

@@ -34,7 +34,6 @@ re_log_types.workspace = true
 re_protos.workspace = true
 re_smart_channel.workspace = true
 re_sorbet.workspace = true
-re_types.workspace = true
 
 arrow.workspace = true
 async-stream.workspace = true

crates/store/re_grpc_client/src/redap/address.rs

@@ -32,15 +32,14 @@ pub enum ConnectionError {
     #[error("unexpected base URL: {0}")]
     UnexpectedBaseUrl(String),
 
-    /// The given url is not a valid Rerun storage node URL.
-    #[error("URL {url:?} should follow rerun://host:port/recording/12345 for recording or rerun://host:port/catalog for catalog")]
-    InvalidAddress { url: String, msg: String },
+    #[error("URL {url:?} cannot be loaded as a recording")]
+    CannotLoadUrlAsRecording { url: String },
 }
 
 /// The different schemes supported by Rerun.
 ///
 /// We support `rerun`, `rerun+http`, and `rerun+https`.
-#[derive(Debug, PartialEq, Eq)]
+#[derive(Debug, PartialEq, Eq, Copy, Clone, Hash)]
 pub enum Scheme {
     Rerun,
     RerunHttp,
@@ -59,15 +58,15 @@ impl std::fmt::Display for Scheme {
 
 impl Scheme {
     /// Converts a [`Scheme`] to either `http` or `https`.
-    fn to_http_scheme(&self) -> &str {
+    fn as_http_scheme(&self) -> &str {
         match self {
             Self::Rerun | Self::RerunHttps => "https",
             Self::RerunHttp => "http",
         }
     }
 }
 
-#[derive(Debug, PartialEq, Eq)]
+#[derive(Debug, PartialEq, Eq, Clone, Hash)]
 pub struct Origin {
     scheme: Scheme,
     host: url::Host<String>,
@@ -82,7 +81,7 @@ impl Origin {
     fn to_http_scheme(&self) -> String {
         format!(
             "{}://{}:{}",
-            self.scheme.to_http_scheme(),
+            self.scheme.as_http_scheme(),
             self.host,
             self.port
         )
@@ -242,9 +241,9 @@ mod tests {
 
     #[test]
     fn scheme_conversion() {
-        assert_eq!(Scheme::Rerun.to_http_scheme(), "https");
-        assert_eq!(Scheme::RerunHttp.to_http_scheme(), "http");
-        assert_eq!(Scheme::RerunHttps.to_http_scheme(), "https");
+        assert_eq!(Scheme::Rerun.as_http_scheme(), "https");
+        assert_eq!(Scheme::RerunHttp.as_http_scheme(), "http");
+        assert_eq!(Scheme::RerunHttps.as_http_scheme(), "https");
     }
 
     #[test]