Remove some dead code from re_redap_client (#11362)

Author: Wumpf

Cargo.lock

@@ -9203,7 +9203,6 @@ dependencies = [
  "re_arrow_util",
  "re_auth",
  "re_chunk",
- "re_error",
  "re_log",
  "re_log_encoding",
  "re_log_types",
@@ -9220,7 +9219,6 @@ dependencies = [
  "tower",
  "tracing",
  "url",
- "wasm-bindgen-futures",
 ]
 
 [[package]]

crates/store/re_redap_client/Cargo.toml

@@ -34,7 +34,6 @@ all-features = true
 re_arrow_util.workspace = true
 re_auth.workspace = true
 re_chunk.workspace = true
-re_error.workspace = true
 re_log.workspace = true
 re_log_encoding = { workspace = true, features = ["encoder", "decoder"] }
 re_log_types.workspace = true
@@ -65,5 +64,4 @@ tonic = { workspace = true, default-features = false, features = [
 # Web dependencies:
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 tonic-web-wasm-client.workspace = true
-wasm-bindgen-futures.workspace = true
 tonic = { workspace = true, default-features = false }

crates/store/re_redap_client/src/grpc.rs

@@ -8,14 +8,11 @@ use re_protos::cloud::v1alpha1::GetChunksRequest;
 use re_protos::cloud::v1alpha1::ext::{Query, QueryLatestAt, QueryRange};
 use re_protos::cloud::v1alpha1::rerun_cloud_service_client::RerunCloudServiceClient;
 use re_protos::common::v1alpha1::ext::PartitionId;
-use re_uri::{DatasetPartitionUri, Origin, TimeSelection};
+use re_uri::{Origin, TimeSelection};
 
 use tokio_stream::{Stream, StreamExt as _};
 
-use crate::{
-    ConnectionClient, ConnectionRegistryHandle, MAX_DECODING_MESSAGE_SIZE, StreamError,
-    StreamPartitionError, spawn_future,
-};
+use crate::{ConnectionClient, MAX_DECODING_MESSAGE_SIZE, StreamError, StreamPartitionError};
 
 /// UI commands issued when streaming in datasets.
 ///
@@ -35,55 +32,6 @@ pub enum UiCommand {
     },
 }
 
-/// Stream an rrd file or metadata catalog over gRPC from a Rerun Data Platform server.
-///
-/// `on_msg` can be used to wake up the UI thread on Wasm.
-pub fn stream_dataset_from_redap(
-    connection_registry: &ConnectionRegistryHandle,
-    uri: DatasetPartitionUri,
-    on_ui_cmd: Option<Box<dyn Fn(UiCommand) + Send + Sync>>,
-    on_msg: Option<Box<dyn Fn() + Send + Sync>>,
-) -> re_smart_channel::Receiver<LogMsg> {
-    re_log::debug!("Loading {uri}…");
-
-    let (tx, rx) = re_smart_channel::smart_channel(
-        re_smart_channel::SmartMessageSource::RedapGrpcStream {
-            uri: uri.clone(),
-            select_when_loaded: true,
-        },
-        re_smart_channel::SmartChannelSource::RedapGrpcStream {
-            uri: uri.clone(),
-            select_when_loaded: true,
-        },
-    );
-
-    async fn stream_partition(
-        connection_registry: ConnectionRegistryHandle,
-        tx: re_smart_channel::Sender<LogMsg>,
-        uri: DatasetPartitionUri,
-        on_ui_cmd: Option<Box<dyn Fn(UiCommand) + Send + Sync>>,
-        on_msg: Option<Box<dyn Fn() + Send + Sync>>,
-    ) -> Result<(), StreamError> {
-        let client = connection_registry.client(uri.origin.clone()).await?;
-
-        stream_blueprint_and_partition_from_server(client, tx, uri.clone(), on_ui_cmd, on_msg).await
-    }
-
-    let connection_registry = connection_registry.clone();
-    spawn_future(async move {
-        if let Err(err) =
-            stream_partition(connection_registry, tx, uri.clone(), on_ui_cmd, on_msg).await
-        {
-            re_log::error!(
-                "Error while streaming {uri}: {}",
-                re_error::format_ref(&err)
-            );
-        }
-    });
-
-    rx
-}
-
 // TODO(ab): do not publish this out of this crate (for now it is still being used by rerun_py
 // the viewer grpc connection). Ideally we'd only publish `ClientConnectionError`.
 #[derive(Debug, thiserror::Error)]
@@ -568,9 +516,20 @@ async fn stream_partition_from_server(
 
     let store_id = store_info.store_id.clone();
 
+    if tx
+        .send(LogMsg::SetStoreInfo(SetStoreInfo {
+            row_id: *re_chunk::RowId::new(),
+            info: store_info,
+        }))
+        .is_err()
+    {
+        re_log::debug!("Receiver disconnected");
+        return Ok(());
+    }
+
     // Send UI commands for recording (as opposed to blueprint) stores.
     if let Some(on_ui_cmd) = on_ui_cmd
-        && store_info.store_id.is_recording()
+        && store_id.is_recording()
     {
         if let Some(time_range) = time_range {
             on_ui_cmd(UiCommand::AddValidTimeRange {
@@ -594,17 +553,6 @@ async fn stream_partition_from_server(
         }
     }
 
-    if tx
-        .send(LogMsg::SetStoreInfo(SetStoreInfo {
-            row_id: *re_chunk::RowId::new(),
-            info: store_info,
-        }))
-        .is_err()
-    {
-        re_log::debug!("Receiver disconnected");
-        return Ok(());
-    }
-
     // TODO(#10229): this looks to be converting back and forth?
 
     let static_chunk_stream = get_chunks_response_to_chunk_and_partition_id(static_chunk_stream);

crates/store/re_redap_client/src/lib.rs

@@ -13,7 +13,6 @@ pub use self::{
         ConnectionError, RedapClient, UiCommand, channel,
         fetch_chunks_response_to_chunk_and_partition_id,
         get_chunks_response_to_chunk_and_partition_id, stream_blueprint_and_partition_from_server,
-        stream_dataset_from_redap,
     },
 };
 
@@ -157,25 +156,3 @@ const _: () = assert!(
     std::mem::size_of::<StreamError>() <= 80,
     "Error type is too large. Try to reduce its size by boxing some of its variants.",
 );
-
-// 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);
-}