Add serve_grpc API

crates/top/re_sdk/Cargo.toml

@@ -46,6 +46,8 @@ web_viewer = [
   "dep:webbrowser",
 ]
 
+server = ["dep:re_smart_channel", "dep:tokio"]
+
 
 [dependencies]
 re_build_info.workspace = true

crates/top/re_sdk/src/grpc_server.rs

@@ -0,0 +1,79 @@
+use re_log_types::LogMsg;
+
+/// A [`crate::sink::LogSink`] tied to a hosted Rerun gRPC server.
+///
+/// The hosted gRPC server may be connected to by any SDK or Viewer.
+///
+/// All data sent through this sink is immediately redirected to the gRPC server.
+pub struct GrpcServerSink {
+    /// Sender to send messages to the gRPC server.
+    sender: re_smart_channel::Sender<LogMsg>,
+
+    /// The gRPC server thread.
+    _server_handle: std::thread::JoinHandle<()>,
+
+    /// Rerun websocket server.
+    server_shutdown_signal: re_grpc_server::shutdown::Signal,
+}
+
+impl GrpcServerSink {
+    /// A `bind_ip` of `"0.0.0.0"` is a good default.
+    pub fn new(
+        bind_ip: &str,
+        grpc_port: u16,
+        server_memory_limit: re_memory::MemoryLimit,
+    ) -> Result<Self, std::net::AddrParseError> {
+        let (server_shutdown_signal, shutdown) = re_grpc_server::shutdown::shutdown();
+
+        let grpc_server_addr = format!("{bind_ip}:{grpc_port}").parse()?;
+        let (channel_tx, channel_rx) = re_smart_channel::smart_channel::<re_log_types::LogMsg>(
+            re_smart_channel::SmartMessageSource::MessageProxy {
+                url: format!("rerun+http://{grpc_server_addr}/proxy"),
+            },
+            re_smart_channel::SmartChannelSource::Sdk,
+        );
+        let server_handle = std::thread::Builder::new()
+            .name("message_proxy_server".to_owned())
+            .spawn(move || {
+                let mut builder = tokio::runtime::Builder::new_current_thread();
+                builder.enable_all();
+                let rt = builder.build().expect("failed to build tokio runtime");
+
+                rt.block_on(re_grpc_server::serve_from_channel(
+                    grpc_server_addr,
+                    server_memory_limit,
+                    shutdown,
+                    channel_rx,
+                ));
+            })
+            .expect("failed to spawn thread for message proxy server");
+
+        Ok(Self {
+            sender: channel_tx,
+            _server_handle: server_handle,
+            server_shutdown_signal,
+        })
+    }
+}
+
+impl crate::sink::LogSink for GrpcServerSink {
+    fn send(&self, msg: LogMsg) {
+        if let Err(err) = self.sender.send(msg) {
+            re_log::error_once!("Failed to send log message to gRPC server: {err}");
+        }
+    }
+
+    #[inline]
+    fn flush_blocking(&self) {
+        if let Err(err) = self.sender.flush_blocking() {
+            re_log::error_once!("Failed to flush: {err}");
+        }
+    }
+}
+
+impl Drop for GrpcServerSink {
+    fn drop(&mut self) {
+        self.sender.flush_blocking().ok();
+        self.server_shutdown_signal.stop();
+    }
+}

crates/top/re_sdk/src/lib.rs

@@ -123,6 +123,10 @@ pub use re_data_loader::{DataLoader, DataLoaderError, DataLoaderSettings, Loaded
 #[cfg(feature = "web_viewer")]
 pub mod web_viewer;
 
+/// Method for spawning a gRPC server and streaming the SDK log stream to it.
+#[cfg(feature = "server")]
+pub mod grpc_server;
+
 /// Re-exports of other crates.
 pub mod external {
     pub use re_grpc_client;

crates/top/re_sdk/src/recording_stream.rs

@@ -97,6 +97,10 @@ pub enum RecordingStreamError {
     /// Invalid endpoint
     #[error("not a `/proxy` endpoint")]
     NotAProxyEndpoint,
+
+    /// Invalid bind IP.
+    #[error(transparent)]
+    InvalidAddress(#[from] std::net::AddrParseError),
 }
 
 /// Results that can occur when creating/manipulating a [`RecordingStream`].
@@ -401,6 +405,63 @@ impl RecordingStreamBuilder {
         }
     }
 
+    #[cfg(feature = "server")]
+    /// Creates a new [`RecordingStream`] that is pre-configured to stream the data through to a
+    /// locally hosted gRPC server.
+    ///
+    /// The server is hosted on the default IP and port, and may be connected to by any SDK or Viewer
+    /// at `rerun+http://127.0.0.1:9876/proxy`.
+    ///
+    /// To configure the gRPC server's IP and port, use [`Self::serve_grpc_opts`] instead.
+    ///
+    /// The gRPC server will buffer in memory so that late connecting viewers will still get all the data.
+    /// You can limit the amount of data buffered by the gRPC server using [`Self::serve_grpc_opts`],
+    /// with the `server_memory_limit` argument. Once the memory limit is reached, the earliest logged data
+    /// will be dropped. Static data is never dropped.
+    pub fn serve_grpc(self) -> RecordingStreamResult<RecordingStream> {
+        self.serve_grpc_opts(
+            "0.0.0.0",
+            9876,
+            re_memory::MemoryLimit::from_fraction_of_total(0.75),
+        )
+    }
+
+    #[cfg(feature = "server")]
+    /// Creates a new [`RecordingStream`] that is pre-configured to stream the data through to a
+    /// locally hosted gRPC server.
+    ///
+    /// The server is hosted on the given `bind_ip` and `port`, may be connected to by any SDK or Viewer
+    /// at `rerun+http://{bind_ip}:{port}/proxy`.
+    ///
+    /// `0.0.0.0` is a good default for `bind_ip`.
+    ///
+    /// The gRPC server will buffer all log data in memory so that late connecting viewers will get all the data.
+    /// You can limit the amount of data buffered by the gRPC server with the `server_memory_limit` argument.
+    /// Once reached, the earliest logged data will be dropped. Static data is never dropped.
+    pub fn serve_grpc_opts(
+        self,
+        bind_ip: impl AsRef<str>,
+        port: u16,
+        server_memory_limit: re_memory::MemoryLimit,
+    ) -> RecordingStreamResult<RecordingStream> {
+        let (enabled, store_info, properties, batcher_config) = self.into_args();
+        if enabled {
+            RecordingStream::new(
+                store_info,
+                properties,
+                batcher_config,
+                Box::new(crate::grpc_server::GrpcServerSink::new(
+                    bind_ip.as_ref(),
+                    port,
+                    server_memory_limit,
+                )?),
+            )
+        } else {
+            re_log::debug!("Rerun disabled - call to serve_grpc() ignored");
+            Ok(RecordingStream::disabled())
+        }
+    }
+
     /// Creates a new [`RecordingStream`] that is pre-configured to stream the data through to an
     /// RRD file on disk.
     ///
@@ -1809,6 +1870,49 @@ impl RecordingStream {
         Ok(())
     }
 
+    #[cfg(feature = "server")]
+    /// Swaps the underlying sink for a [`crate::grpc_server::GrpcServerSink`] pre-configured to listen on
+    /// `rerun+http://127.0.0.1:9876/proxy`.
+    ///
+    /// To configure the gRPC server's IP and port, use [`Self::serve_grpc_opts`] instead.
+    ///
+    /// The gRPC server will buffer all log data in memory so that late connecting viewers will get all the data.
+    /// You can limit the amount of data buffered by the gRPC server with the `server_memory_limit` argument.
+    /// Once reached, the earliest logged data will be dropped. Static data is never dropped.
+    pub fn serve_grpc(
+        &self,
+        server_memory_limit: re_memory::MemoryLimit,
+    ) -> RecordingStreamResult<()> {
+        self.serve_grpc_opts("0.0.0.0", 9876, server_memory_limit)
+    }
+
+    #[cfg(feature = "server")]
+    /// Swaps the underlying sink for a [`crate::grpc_server::GrpcServerSink`] pre-configured to listen on
+    /// `rerun+http://{bind_ip}:{port}/proxy`.
+    ///
+    /// `0.0.0.0` is a good default for `bind_ip`.
+    ///
+    /// The gRPC server will buffer all log data in memory so that late connecting viewers will get all the data.
+    /// You can limit the amount of data buffered by the gRPC server with the `server_memory_limit` argument.
+    /// Once reached, the earliest logged data will be dropped. Static data is never dropped.
+    pub fn serve_grpc_opts(
+        &self,
+        bind_ip: impl AsRef<str>,
+        port: u16,
+        server_memory_limit: re_memory::MemoryLimit,
+    ) -> RecordingStreamResult<()> {
+        if forced_sink_path().is_some() {
+            re_log::debug!("Ignored setting GrpcServerSink since {ENV_FORCE_SAVE} is set");
+            return Ok(());
+        }
+
+        let sink =
+            crate::grpc_server::GrpcServerSink::new(bind_ip.as_ref(), port, server_memory_limit)?;
+
+        self.set_sink(Box::new(sink));
+        Ok(())
+    }
+
     /// Spawns a new Rerun Viewer process from an executable available in PATH, then swaps the
     /// underlying sink for a [`crate::log_sink::GrpcSink`] sink pre-configured to send data to that
     /// new process.

crates/top/rerun/Cargo.toml

@@ -101,7 +101,7 @@ run = [
 ]
 
 ## Support for running a gRPC server that listens to incoming log messages from a Rerun SDK.
-server = ["dep:re_grpc_server"]
+server = ["dep:re_grpc_server", "re_sdk/server", "tokio/signal"]
 
 ## Embed the Rerun SDK & built-in types and re-export all of their public symbols.
 sdk = ["dep:re_sdk", "dep:re_types"]

crates/top/rerun/src/commands/entrypoint.rs

@@ -57,12 +57,18 @@ Examples:
     Open an .rrd file and stream it to a Web Viewer:
         rerun recording.rrd --web-viewer
 
-    Host a Rerun gRPC server which listens for incoming connections from the logging SDK, buffer the log messages, and serves the results:
+    Host a Rerun gRPC server which listens for incoming connections from the logging SDK, buffers the log messages, and serves the results:
         rerun --serve-web
 
     Host a Rerun Server which serves a recording from a file over gRPC to any connecting Rerun Viewers:
         rerun --serve-web recording.rrd
 
+    Host a Rerun gRPC server without spawning a Viewer:
+        rerun --serve-grpc
+
+    Spawn a Viewer without also hosting a gRPC server:
+        rerun --connect
+
     Connect to a Rerun Server:
         rerun rerun+http://localhost:9877/proxy
 
@@ -158,6 +164,13 @@ When persisted, the state will be stored at the following locations:
     #[clap(long)]
     serve_web: bool,
 
+    /// This will host a gRPC server.
+    ///
+    /// The server will act like a proxy, listening for incoming connections from
+    /// logging SDKs, and forwarding it to Rerun viewers.
+    #[clap(long)]
+    serve_grpc: bool,
+
     /// Do not attempt to start a new server, instead try to connect to an existing one.
     ///
     /// Optionally accepts an HTTP(S) URL to a gRPC server.
@@ -638,7 +651,7 @@ fn run_impl(
     _build_info: re_build_info::BuildInfo,
     call_source: CallSource,
     args: Args,
-    _tokio_runtime_handle: &tokio::runtime::Handle,
+    tokio_runtime_handle: &tokio::runtime::Handle,
 ) -> anyhow::Result<()> {
     #[cfg(feature = "native_viewer")]
     let profiler = run_profiler(&args);
@@ -660,8 +673,11 @@ fn run_impl(
         re_viewer::StartupOptions {
             hide_welcome_screen: args.hide_welcome_screen,
             detach_process: args.detach_process,
-            memory_limit: re_memory::MemoryLimit::parse(&args.memory_limit)
-                .map_err(|err| anyhow::format_err!("Bad --memory-limit: {err}"))?,
+            memory_limit: {
+                re_log::debug!("Parsing memory limit for Viewer");
+                re_memory::MemoryLimit::parse(&args.memory_limit)
+                    .map_err(|err| anyhow::format_err!("Bad --memory-limit: {err}"))?
+            },
             persist_state: args.persist_state,
             is_in_notebook: false,
             screenshot_to_path_then_quit: args.screenshot_to.clone(),
@@ -690,8 +706,11 @@ fn run_impl(
     #[cfg(feature = "server")]
     let server_addr = std::net::SocketAddr::new(args.bind, args.port);
     #[cfg(feature = "server")]
-    let server_memory_limit = re_memory::MemoryLimit::parse(&args.server_memory_limit)
-        .map_err(|err| anyhow::format_err!("Bad --server-memory-limit: {err}"))?;
+    let server_memory_limit = {
+        re_log::debug!("Parsing memory limit for gRPC server");
+        re_memory::MemoryLimit::parse(&args.server_memory_limit)
+            .map_err(|err| anyhow::format_err!("Bad --server-memory-limit: {err}"))?
+    };
 
     #[allow(unused_variables)]
     let (command_sender, command_receiver) = re_viewer_context::command_channel();
@@ -782,7 +801,7 @@ fn run_impl(
             //       we want all receivers to push their data to the server.
             //       For that we spawn the server a bit further down, after we've collected
             //       all receivers into `rxs`.
-            } else if !args.serve && !args.serve_web {
+            } else if !args.serve && !args.serve_web && !args.serve_grpc {
                 let server: Receiver<LogMsg> = re_grpc_server::spawn_with_recv(
                     server_addr,
                     server_memory_limit,
@@ -811,6 +830,35 @@ fn run_impl(
 
         let rx = ReceiveSet::new(rxs);
         Ok(stream_to_rrd_on_disk(&rx, &rrd_path.into())?)
+    } else if args.serve_grpc {
+        if !catalog_endpoints.is_empty() {
+            anyhow::bail!("`--serve` does not support catalogs");
+        }
+
+        if !cfg!(feature = "server") {
+            _ = (call_source, rxs);
+            anyhow::bail!("Can't host server - rerun was not compiled with the 'server' feature");
+        }
+
+        #[cfg(feature = "server")]
+        {
+            let (signal, shutdown) = re_grpc_server::shutdown::shutdown();
+            // Spawn a server which the Web Viewer can connect to.
+            // All `rxs` are consumed by the server.
+            re_grpc_server::spawn_from_rx_set(
+                server_addr,
+                server_memory_limit,
+                shutdown,
+                ReceiveSet::new(rxs),
+            );
+
+            // Gracefully shut down the server on SIGINT
+            tokio_runtime_handle.block_on(tokio::signal::ctrl_c()).ok();
+
+            signal.stop();
+        }
+
+        Ok(())
     } else if args.serve || args.serve_web {
         if !catalog_endpoints.is_empty() {
             anyhow::bail!("`--serve` does not support catalogs");
@@ -917,7 +965,7 @@ fn run_impl(
     } else {
         #[cfg(feature = "native_viewer")]
         {
-            let tokio_runtime_handle = _tokio_runtime_handle.clone();
+            let tokio_runtime_handle = tokio_runtime_handle.clone();
 
             return re_viewer::run_native_app(
                 _main_thread_token,

crates/top/rerun_c/Cargo.toml

@@ -37,7 +37,7 @@ test = false
 [dependencies]
 re_arrow_util.workspace = true
 re_log = { workspace = true, features = ["setup"] }
-re_sdk = { workspace = true, features = ["data_loaders"] }
+re_sdk = { workspace = true, features = ["data_loaders", "server"] }
 re_video.workspace = true
 
 ahash.workspace = true