Add and use EndpointAddr for URI structs

Signed-off-by: methylDragon <methylDragon@intrinsic.ai>

Author: methylDragon

crates/store/re_data_source/src/data_source.rs

@@ -252,7 +252,7 @@ impl LogDataSource {
                 let uri_clone = uri.clone();
                 let stream_partition = async move {
                     let client = connection_registry
-                        .client(uri_clone.origin.clone())
+                        .client(uri_clone.endpoint_addr.origin.clone())
                         .await
                         .map_err(|err| ApiError::connection(err, "failed to connect to server"))?;
                     re_redap_client::stream_blueprint_and_partition_from_server(

crates/store/re_grpc_client/src/read.rs

@@ -39,7 +39,7 @@ async fn stream_async(
     on_msg: Option<Box<dyn Fn() + Send + Sync>>,
 ) -> Result<(), StreamError> {
     let mut client = {
-        let url = uri.origin.as_url();
+        let url = uri.endpoint_addr.origin.as_url();
 
         #[cfg(target_arch = "wasm32")]
         let tonic_client = {

crates/store/re_grpc_client/src/write.rs

@@ -314,7 +314,7 @@ async fn message_proxy_client(
     compression: Compression,
     status: Arc<AtomicCell<ClientConnectionState>>,
 ) {
-    let endpoint = match Endpoint::from_shared(uri.origin.as_url()) {
+    let endpoint = match Endpoint::from_shared(uri.endpoint_addr.origin.as_url()) {
         Ok(endpoint) => endpoint,
         Err(err) => {
             status.store(ClientConnectionState::Disconnected(Err(

crates/store/re_grpc_server/src/lib.rs

@@ -360,10 +360,9 @@ pub fn spawn_with_recv(
     re_smart_channel::Receiver<re_log_types::DataSourceMessage>,
     crossbeam::channel::Receiver<re_log_types::TableMsg>,
 ) {
-    let uri = re_uri::ProxyUri::new(
+    let uri = re_uri::ProxyUri::new(re_uri::EndpointAddr::new(
         re_uri::Origin::from_scheme_and_socket_addr(re_uri::Scheme::RerunHttp, addr),
-        String::new(),
-    );
+    ));
     let (channel_log_tx, channel_log_rx) = re_smart_channel::smart_channel(
         re_smart_channel::SmartMessageSource::MessageProxy(uri.clone()),
         re_smart_channel::SmartChannelSource::MessageProxy(uri),

crates/store/re_redap_client/src/grpc.rs

@@ -350,8 +350,7 @@ pub async fn stream_blueprint_and_partition_from_server(
     }
 
     let re_uri::DatasetPartitionUri {
-        origin: _,
-        prefix: _,
+        endpoint_addr: _,
         dataset_id,
         partition_id,
         time_range,

crates/top/re_sdk/src/grpc_server.rs

@@ -34,13 +34,12 @@ impl GrpcServerSink {
 
         let grpc_server_addr = format!("{bind_ip}:{grpc_port}").parse()?;
 
-        let uri = re_uri::ProxyUri::new(
+        let uri = re_uri::ProxyUri::new(re_uri::EndpointAddr::new(
             re_uri::Origin::from_scheme_and_socket_addr(
                 re_uri::Scheme::RerunHttp,
                 grpc_server_addr,
             ),
-            String::new(),
-        );
+        ));
         let (channel_tx, channel_rx) = re_smart_channel::smart_channel::<re_log_types::LogMsg>(
             re_smart_channel::SmartMessageSource::MessageProxy(uri.clone()),
             re_smart_channel::SmartChannelSource::Sdk,

crates/top/re_sdk/src/web_viewer.rs

@@ -49,13 +49,12 @@ impl WebViewerSink {
         let (server_shutdown_signal, shutdown) = re_grpc_server::shutdown::shutdown();
 
         let grpc_server_addr = format!("{bind_ip}:{grpc_port}").parse()?;
-        let uri = re_uri::ProxyUri::new(
+        let uri = re_uri::ProxyUri::new(re_uri::EndpointAddr::new(
             re_uri::Origin::from_scheme_and_socket_addr(
                 re_uri::Scheme::RerunHttp,
                 grpc_server_addr,
             ),
-            String::new(),
-        );
+        ));
         let (channel_tx, channel_rx) = re_smart_channel::smart_channel::<re_log_types::LogMsg>(
             re_smart_channel::SmartMessageSource::MessageProxy(uri),
             re_smart_channel::SmartChannelSource::Sdk,

crates/utils/re_uri/src/endpoint_addr.rs

@@ -0,0 +1,102 @@
+use crate::Origin;
+
+/// A Rerun endpoint address, consisting of an origin and an optional path prefix.
+///
+/// Example: `https://rerun.io:443/custom/prefix`
+#[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
+pub struct EndpointAddr {
+    pub origin: Origin,
+
+    /// An optional path prefix, e.g. `/my/prefix`.
+    ///
+    /// The prefix is guaranteed to start with a slash if it is not empty,
+    /// and guaranteed not to end with a slash.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub path_prefix: Option<String>,
+}
+
+impl EndpointAddr {
+    /// Create a new [`EndpointAddr`] with the given origin and no path prefix.
+    pub fn new(origin: Origin) -> Self {
+        Self {
+            origin,
+            path_prefix: None,
+        }
+    }
+
+    /// Add a path prefix to the endpoint address.
+    ///
+    /// The prefix is normalized:
+    /// - It will be ensured to start with a `/` if not empty.
+    /// - Trailing slashes will be removed.
+    pub fn with_path_prefix(mut self, path_prefix: impl Into<String>) -> Self {
+        let path_prefix = path_prefix.into();
+        if path_prefix.is_empty() || path_prefix == "/" {
+            self.path_prefix = None;
+            return self;
+        }
+
+        let mut path_prefix = path_prefix.trim_end_matches('/').to_owned();
+        if !path_prefix.starts_with('/') {
+            path_prefix.insert(0, '/');
+        }
+
+        self.path_prefix = Some(path_prefix);
+        self
+    }
+}
+
+impl std::fmt::Display for EndpointAddr {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let Self {
+            origin,
+            path_prefix,
+        } = self;
+
+        write!(f, "{origin}")?;
+        if let Some(prefix) = path_prefix {
+            write!(f, "{prefix}")?;
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::Scheme;
+
+    fn test_origin() -> Origin {
+        Origin::from_scheme_and_socket_addr(Scheme::Rerun, "127.0.0.1:1234".parse().unwrap())
+    }
+
+    #[test]
+    fn test_serialization() {
+        let origin = test_origin();
+        let addr = EndpointAddr::new(origin.clone());
+        assert_eq!(addr.path_prefix, None);
+        assert_eq!(addr.to_string(), "rerun://127.0.0.1:1234");
+
+        let addr = EndpointAddr::new(origin.clone()).with_path_prefix("");
+        assert_eq!(addr.path_prefix, None);
+
+        let addr = EndpointAddr::new(origin.clone()).with_path_prefix("/");
+        assert_eq!(addr.path_prefix, None);
+
+        let addr = EndpointAddr::new(origin.clone()).with_path_prefix("foo");
+        assert_eq!(addr.path_prefix.as_deref(), Some("/foo"));
+        assert_eq!(addr.to_string(), "rerun://127.0.0.1:1234/foo");
+
+        let addr = EndpointAddr::new(origin.clone()).with_path_prefix("/foo");
+        assert_eq!(addr.path_prefix.as_deref(), Some("/foo"));
+        assert_eq!(addr.to_string(), "rerun://127.0.0.1:1234/foo");
+
+        let addr = EndpointAddr::new(origin.clone()).with_path_prefix("foo/");
+        assert_eq!(addr.path_prefix.as_deref(), Some("/foo"));
+        assert_eq!(addr.to_string(), "rerun://127.0.0.1:1234/foo");
+
+        let addr = EndpointAddr::new(origin.clone()).with_path_prefix("/foo/bar");
+        assert_eq!(addr.path_prefix.as_deref(), Some("/foo/bar"));
+        assert_eq!(addr.to_string(), "rerun://127.0.0.1:1234/foo/bar");
+    }
+}

crates/utils/re_uri/src/endpoints/catalog.rs

@@ -1,21 +1,21 @@
-use crate::{Origin, RedapUri};
+use crate::{EndpointAddr, RedapUri};
 
 /// `scheme://hostname:port/catalog`
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct CatalogUri {
-    pub origin: Origin,
-    pub prefix: String,
+    pub endpoint_addr: EndpointAddr,
 }
 
 impl std::fmt::Display for CatalogUri {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}{}/catalog", self.origin, self.prefix)
+        let Self { endpoint_addr } = self;
+        write!(f, "{endpoint_addr}/catalog")
     }
 }
 
 impl CatalogUri {
-    pub fn new(origin: Origin, prefix: String) -> Self {
-        Self { origin, prefix }
+    pub fn new(endpoint_addr: EndpointAddr) -> Self {
+        Self { endpoint_addr }
     }
 }
 

crates/utils/re_uri/src/endpoints/dataset.rs

@@ -1,6 +1,6 @@
 use re_log_types::StoreId;
 
-use crate::{Error, Fragment, Origin, RedapUri, TimeSelection};
+use crate::{EndpointAddr, Error, Fragment, RedapUri, TimeSelection};
 
 /// URI pointing at the data underlying a dataset.
 ///
@@ -11,8 +11,7 @@ use crate::{Error, Fragment, Origin, RedapUri, TimeSelection};
 /// In the future we will add richer queries.
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct DatasetPartitionUri {
-    pub origin: Origin,
-    pub prefix: String,
+    pub endpoint_addr: EndpointAddr,
     pub dataset_id: re_tuid::Tuid,
 
     // Query parameters: these affect what data is returned.
@@ -26,22 +25,26 @@ pub struct DatasetPartitionUri {
 
 impl std::fmt::Display for DatasetPartitionUri {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "{}{}/dataset/{}",
-            self.origin, self.prefix, self.dataset_id
-        )?;
+        let Self {
+            endpoint_addr,
+            dataset_id,
+            partition_id,
+            time_range,
+            fragment,
+        } = self;
+
+        write!(f, "{endpoint_addr}/dataset/{dataset_id}")?;
 
         // ?query:
         {
-            write!(f, "?partition_id={}", self.partition_id)?;
+            write!(f, "?partition_id={partition_id}")?;
         }
-        if let Some(time_range) = &self.time_range {
+        if let Some(time_range) = time_range {
             write!(f, "&time_range={time_range}")?;
         }
 
         // #fragment:
-        let fragment = self.fragment.to_string();
+        let fragment = fragment.to_string();
         if !fragment.is_empty() {
             write!(f, "#{fragment}")?;
         }
@@ -52,8 +55,7 @@ impl std::fmt::Display for DatasetPartitionUri {
 
 impl DatasetPartitionUri {
     pub fn new(
-        origin: Origin,
-        prefix: String,
+        endpoint_addr: EndpointAddr,
         dataset_id: re_tuid::Tuid,
         url: &url::Url,
     ) -> Result<Self, Error> {
@@ -87,8 +89,7 @@ impl DatasetPartitionUri {
         }
 
         Ok(Self {
-            origin,
-            prefix,
+            endpoint_addr,
             dataset_id,
             partition_id,
             time_range,
@@ -99,10 +100,9 @@ impl DatasetPartitionUri {
     /// Returns [`Self`] without any (optional) `?query` or `#fragment`.
     pub fn without_query_and_fragment(mut self) -> Self {
         let Self {
-            origin: _,       // Mandatory
-            prefix: _,       // Mandatory
-            dataset_id: _,   // Mandatory
-            partition_id: _, // Mandatory
+            endpoint_addr: _, // Mandatory
+            dataset_id: _,    // Mandatory
+            partition_id: _,  // Mandatory
             time_range,
             fragment,
         } = &mut self;
@@ -116,10 +116,9 @@ impl DatasetPartitionUri {
     /// Returns [`Self`] without any (optional) `#fragment`.
     pub fn without_fragment(mut self) -> Self {
         let Self {
-            origin: _,       // Mandatory
-            prefix: _,       // Mandatory
-            dataset_id: _,   // Mandatory
-            partition_id: _, // Mandatory
+            endpoint_addr: _, // Mandatory
+            dataset_id: _,    // Mandatory
+            partition_id: _,  // Mandatory
             time_range: _,
             fragment,
         } = &mut self;