say goodbye when gracefully disconnecting

crates/re_sdk/src/recording_stream.rs

@@ -339,12 +339,6 @@ impl Drop for RecordingStreamInner {
         // sending data down the pipeline.
         self.batcher.flush_blocking();
         self.cmds_tx.send(Command::PopPendingTables).ok();
-        self.cmds_tx
-            .send(Command::RecordMsg(LogMsg::Goodbye(
-                self.info.recording_id.clone(),
-                RowId::random(),
-            )))
-            .ok();
         self.cmds_tx.send(Command::Shutdown).ok();
         if let Some(handle) = self.batcher_to_sink_handle.take() {
             handle.join().ok();
@@ -771,12 +765,25 @@ impl RecordingStream {
         Ok(())
     }
 
-    /// Swaps the underlying sink for a [`crate::sink::BufferedSink`].
+    /// Swaps the underlying sink for a [`crate::sink::BufferedSink`], making sure to first send a
+    /// `Goodbye` message down the sink to let know the other end of the graceful disconnection.
     ///
     /// 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 disconnect(&self) {
+        let Some(this) = &*self.inner else {
+            re_log::warn_once!("Recording disabled - call to disconnect() ignored");
+            return;
+        };
+
+        this.cmds_tx
+            .send(Command::RecordMsg(LogMsg::Goodbye(
+                this.info.recording_id.clone(),
+                RowId::random(),
+            )))
+            .ok();
+
         self.set_sink(Box::new(crate::sink::BufferedSink::new()));
     }
 }