RecordingStream: introduce connect_opts (#4042)

Promised follow up after the addition of `spawn_opts`.

This simplifies a bunch of things and greatly improve DX while we wait
for Rust to unlock the default parameter skill ✨

crates/re_sdk/src/recording_stream.rs

@@ -255,6 +255,21 @@ impl RecordingStreamBuilder {
         }
     }
 
+    /// Creates a new [`RecordingStream`] that is pre-configured to stream the data through to a
+    /// remote Rerun instance.
+    ///
+    /// See also [`Self::connect_opts`] if you wish to configure the TCP connection.
+    ///
+    /// ## Example
+    ///
+    /// ```no_run
+    /// let rec = re_sdk::RecordingStreamBuilder::new("rerun_example_app").connect()?;
+    /// # Ok::<(), Box<dyn std::error::Error>>(())
+    /// ```
+    pub fn connect(self) -> RecordingStreamResult<RecordingStream> {
+        self.connect_opts(crate::default_server_addr(), crate::default_flush_timeout())
+    }
+
     /// Creates a new [`RecordingStream`] that is pre-configured to stream the data through to a
     /// remote Rerun instance.
     ///
@@ -266,10 +281,10 @@ impl RecordingStreamBuilder {
     ///
     /// ```no_run
     /// let rec = re_sdk::RecordingStreamBuilder::new("rerun_example_app")
-    ///     .connect(re_sdk::default_server_addr(), re_sdk::default_flush_timeout())?;
+    ///     .connect_opts(re_sdk::default_server_addr(), re_sdk::default_flush_timeout())?;
     /// # Ok::<(), Box<dyn std::error::Error>>(())
     /// ```
-    pub fn connect(
+    pub fn connect_opts(
         self,
         addr: std::net::SocketAddr,
         flush_timeout: Option<std::time::Duration>,
@@ -321,21 +336,17 @@ impl RecordingStreamBuilder {
     /// If a Rerun Viewer is already listening on this TCP port, the stream will be redirected to
     /// that viewer instead of starting a new one.
     ///
-    /// `flush_timeout` is the minimum time the [`TcpSink`][`crate::log_sink::TcpSink`] will
-    /// wait during a flush before potentially dropping data.  Note: Passing `None` here can cause a
-    /// call to `flush` to block indefinitely if a connection cannot be established.
+    /// See also [`Self::spawn_opts`] if you wish to configure the behavior of thew Rerun process
+    /// as well as the underlying TCP connection.
     ///
     /// ## Example
     ///
     /// ```no_run
-    /// let rec = re_sdk::RecordingStreamBuilder::new("rerun_example_app").spawn(re_sdk::default_flush_timeout())?;
+    /// let rec = re_sdk::RecordingStreamBuilder::new("rerun_example_app").spawn()?;
     /// # Ok::<(), Box<dyn std::error::Error>>(())
     /// ```
-    pub fn spawn(
-        self,
-        flush_timeout: Option<std::time::Duration>,
-    ) -> RecordingStreamResult<RecordingStream> {
-        self.spawn_opts(&Default::default(), flush_timeout)
+    pub fn spawn(self) -> RecordingStreamResult<RecordingStream> {
+        self.spawn_opts(&Default::default(), crate::default_flush_timeout())
     }
 
     /// Spawns a new Rerun Viewer process from an executable available in PATH, then creates a new
@@ -368,12 +379,12 @@ impl RecordingStreamBuilder {
         // NOTE: If `_RERUN_TEST_FORCE_SAVE` is set, all recording streams will write to disk no matter
         // what, thus spawning a viewer is pointless (and probably not intended).
         if forced_sink_path().is_some() {
-            return self.connect(connect_addr, flush_timeout);
+            return self.connect_opts(connect_addr, flush_timeout);
         }
 
         spawn(opts)?;
 
-        self.connect(connect_addr, flush_timeout)
+        self.connect_opts(connect_addr, flush_timeout)
     }
 
     /// Creates a new [`RecordingStream`] that is pre-configured to stream the data through to a
@@ -1150,6 +1161,18 @@ impl RecordingStream {
 }
 
 impl RecordingStream {
+    /// Swaps the underlying sink for a [`crate::log_sink::TcpSink`] sink pre-configured to use
+    /// the specified address.
+    ///
+    /// See also [`Self::connect_opts`] if you wish to configure the TCP connection.
+    ///
+    /// This is a convenience wrapper for [`Self::set_sink`] that upholds the same guarantees in
+    /// terms of data durability and ordering.
+    /// See [`Self::set_sink`] for more information.
+    pub fn connect(&self) {
+        self.connect_opts(crate::default_server_addr(), crate::default_flush_timeout());
+    }
+
     /// Swaps the underlying sink for a [`crate::log_sink::TcpSink`] sink pre-configured to use
     /// the specified address.
     ///
@@ -1160,7 +1183,11 @@ impl RecordingStream {
     /// This is a convenience wrapper for [`Self::set_sink`] that upholds the same guarantees in
     /// terms of data durability and ordering.
     /// See [`Self::set_sink`] for more information.
-    pub fn connect(&self, addr: std::net::SocketAddr, flush_timeout: Option<std::time::Duration>) {
+    pub fn connect_opts(
+        &self,
+        addr: std::net::SocketAddr,
+        flush_timeout: Option<std::time::Duration>,
+    ) {
         if forced_sink_path().is_some() {
             re_log::debug!("Ignored setting new TcpSink since _RERUN_FORCE_SINK is set");
             return;
@@ -1176,15 +1203,14 @@ impl RecordingStream {
     /// If a Rerun Viewer is already listening on this TCP port, the stream will be redirected to
     /// that viewer instead of starting a new one.
     ///
-    /// `flush_timeout` is the minimum time the [`TcpSink`][`crate::log_sink::TcpSink`] will
-    /// wait during a flush before potentially dropping data.  Note: Passing `None` here can cause a
-    /// call to `flush` to block indefinitely if a connection cannot be established.
+    /// See also [`Self::spawn_opts`] if you wish to configure the behavior of thew Rerun process
+    /// as well as the underlying TCP connection.
     ///
     /// This is a convenience wrapper for [`Self::set_sink`] that upholds the same guarantees in
     /// terms of data durability and ordering.
     /// See [`Self::set_sink`] for more information.
-    pub fn spawn(&self, flush_timeout: Option<std::time::Duration>) -> RecordingStreamResult<()> {
-        self.spawn_opts(&Default::default(), flush_timeout)
+    pub fn spawn(&self) -> RecordingStreamResult<()> {
+        self.spawn_opts(&Default::default(), crate::default_flush_timeout())
     }
 
     /// Spawns a new Rerun Viewer process from an executable available in PATH, then swaps the
@@ -1216,7 +1242,7 @@ impl RecordingStream {
 
         spawn(opts)?;
 
-        self.connect(opts.connect_addr(), flush_timeout);
+        self.connect_opts(opts.connect_addr(), flush_timeout);
 
         Ok(())
     }

crates/rerun/src/clap.rs

@@ -91,7 +91,7 @@ impl RerunArgs {
         match self.to_behavior()? {
             RerunBehavior::Connect(addr) => Ok((
                 RecordingStreamBuilder::new(application_id)
-                    .connect(addr, crate::default_flush_timeout())?,
+                    .connect_opts(addr, crate::default_flush_timeout())?,
                 Default::default(),
             )),
 
@@ -101,7 +101,7 @@ impl RerunArgs {
             )),
 
             RerunBehavior::Spawn => Ok((
-                RecordingStreamBuilder::new(application_id).spawn(crate::default_flush_timeout())?,
+                RecordingStreamBuilder::new(application_id).spawn()?,
                 Default::default(),
             )),
 

crates/rerun/src/lib.rs

@@ -55,8 +55,7 @@
 //! # fn positions() -> Vec<rerun::Position3D> { Default::default() }
 //! # fn colors() -> Vec<rerun::Color> { Default::default() }
 //! // Stream log data to an awaiting `rerun` process.
-//! let rec = rerun::RecordingStreamBuilder::new("rerun_example_app")
-//!     .connect(rerun::default_server_addr(), rerun::default_flush_timeout())?;
+//! let rec = rerun::RecordingStreamBuilder::new("rerun_example_app").connect()?;
 //!
 //! let points: Vec<rerun::Position3D> = positions();
 //! let colors: Vec<rerun::Color> = colors();