ref(client): Create EnvelopeSender to wrap Transport

sentry-core/src/batcher.rs

@@ -4,7 +4,7 @@ use std::sync::{Arc, Condvar, Mutex, MutexGuard};
 use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
 
-use crate::client::TransportArc;
+use crate::client::EnvelopeSender;
 use crate::protocol::EnvelopeItem;
 use crate::Envelope;
 use sentry_types::protocol::v7::Log;
@@ -50,7 +50,7 @@ impl Batch for Metric {
 /// Accumulates items in the queue and submits them through the transport when one of the flushing
 /// conditions is met.
 pub(crate) struct Batcher<T: Batch> {
-    transport: TransportArc,
+    envelope_sender: EnvelopeSender,
     queue: Arc<Mutex<BatchQueue<T>>>,
     shutdown: Arc<(Mutex<bool>, Condvar)>,
     worker: Option<JoinHandle<()>>,
@@ -60,13 +60,13 @@ impl<T> Batcher<T>
 where
     T: Batch + Send + 'static,
 {
-    /// Creates a new Batcher that will submit envelopes to the given `transport`.
-    pub(crate) fn new(transport: TransportArc) -> Self {
+    /// Creates a new Batcher that will submit envelopes to the transport.
+    pub(crate) fn new(envelope_sender: EnvelopeSender) -> Self {
         let queue = Arc::new(Mutex::new(BatchQueue { items: Vec::new() }));
         #[allow(clippy::mutex_atomic)]
         let shutdown = Arc::new((Mutex::new(false), Condvar::new()));
 
-        let worker_transport = transport.clone();
+        let worker_envelope_sender = envelope_sender.clone();
         let worker_queue = queue.clone();
         let worker_shutdown = shutdown.clone();
         let worker = std::thread::Builder::new()
@@ -90,7 +90,7 @@ where
                     if last_flush.elapsed() >= FLUSH_INTERVAL {
                         Batcher::flush_queue_internal(
                             worker_queue.lock().unwrap(),
-                            &worker_transport,
+                            &worker_envelope_sender,
                         );
                         last_flush = Instant::now();
                     }
@@ -99,7 +99,7 @@ where
             .unwrap();
 
         Self {
-            transport,
+            envelope_sender,
             queue,
             shutdown,
             worker: Some(worker),
@@ -115,21 +115,24 @@ impl<T: Batch> Batcher<T> {
         let mut queue = self.queue.lock().unwrap();
         queue.items.push(item);
         if queue.items.len() >= MAX_ITEMS {
-            Batcher::flush_queue_internal(queue, &self.transport);
+            Batcher::flush_queue_internal(queue, &self.envelope_sender);
         }
     }
 
     /// Flushes the queue to the transport.
     pub(crate) fn flush(&self) {
         let queue = self.queue.lock().unwrap();
-        Batcher::flush_queue_internal(queue, &self.transport);
+        Batcher::flush_queue_internal(queue, &self.envelope_sender);
     }
 
     /// Flushes the queue to the transport.
     ///
     /// This is a static method as it will be called from both the background
     /// thread and the main thread on drop.
-    fn flush_queue_internal(mut queue_lock: MutexGuard<BatchQueue<T>>, transport: &TransportArc) {
+    fn flush_queue_internal(
+        mut queue_lock: MutexGuard<BatchQueue<T>>,
+        envelope_sender: &EnvelopeSender,
+    ) {
         let items = std::mem::take(&mut queue_lock.items);
         drop(queue_lock);
 
@@ -139,12 +142,10 @@ impl<T: Batch> Batcher<T> {
 
         sentry_debug!("[Batcher({})] Flushing {} items", T::TYPE_NAME, items.len());
 
-        if let Some(ref transport) = *transport.read().unwrap() {
-            let mut envelope = Envelope::new();
-            let envelope_item = T::into_envelope_item(items);
-            envelope.add_item(envelope_item);
-            transport.send_envelope(envelope);
-        }
+        let mut envelope = Envelope::new();
+        let envelope_item = T::into_envelope_item(items);
+        envelope.add_item(envelope_item);
+        envelope_sender.send_envelope(envelope);
     }
 }
 
@@ -157,7 +158,7 @@ impl<T: Batch> Drop for Batcher<T> {
         if let Some(worker) = self.worker.take() {
             worker.join().ok();
         }
-        Batcher::flush_queue_internal(self.queue.lock().unwrap(), &self.transport);
+        Batcher::flush_queue_internal(self.queue.lock().unwrap(), &self.envelope_sender);
     }
 }
 

sentry-core/src/client/envelope_sender.rs

@@ -0,0 +1,187 @@
+//! Contains abstractions for sending envelopes.
+//!
+//! The most important type here is the [`EnvelopeSender`] struct, which wraps a [`Transport`] and
+//! centralizes envelope sending logic. All code in this crate should send envelopes via the
+//! [`EnvelopeSender`], not by using the [`Transport`] directly.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use self::slot::TransportSlot;
+use crate::{Envelope, Transport};
+
+/// Sends envelopes through the client's transport.
+///
+/// Cloning this sender has `Arc`-like semantics: clones share the same transport
+/// slot and send to the same underlying transport until it is shut down.
+///
+/// The [`Default`] implementation creates an [`EnvelopeSender`] without an underlying transport,
+/// effectively rendering calls a no-op.
+#[derive(Clone, Default)]
+pub(crate) struct EnvelopeSender {
+    transport_slot: TransportSlot<dyn Transport>,
+}
+
+impl EnvelopeSender {
+    /// Sends an envelope if the transport is still available.
+    pub(crate) fn send_envelope(&self, envelope: Envelope) {
+        // This forwards to `send_envelope_with`; any envelope pre-processing should be
+        // centralized in the `send_envelope_with` function!
+        self.send_envelope_with(|| Some(envelope));
+    }
+
+    /// Builds and sends an envelope if the transport is still available.
+    ///
+    /// The builder is only executed if this sender is still active. This allows skipping over
+    /// logic that constructs the envelope when it cannot be sent. The builder can also return
+    /// [`None`], in which case, we don't send anything.
+    pub(super) fn send_envelope_with<F>(&self, builder: F)
+    where
+        F: FnOnce() -> Option<Envelope>,
+    {
+        self.transport_slot.send_envelope_with(builder)
+    }
+
+    /// Creates a sender using the transport returned by the provided builder callback.
+    pub(super) fn new<F>(transport_builder: F) -> Self
+    where
+        F: FnOnce() -> Arc<dyn Transport>,
+    {
+        let transport_slot = TransportSlot::new(transport_builder());
+        Self { transport_slot }
+    }
+
+    /// Flushes the transport if it is still available.
+    pub(super) fn flush(&self, timeout: Duration) -> bool {
+        self.transport_slot.flush(timeout)
+    }
+
+    /// Shuts down and removes the transport if it is still available.
+    pub(super) fn shutdown(&self, timeout: Duration) -> bool {
+        self.transport_slot.shutdown(timeout)
+    }
+
+    pub(super) fn clone_with_new_transport_slot(&self) -> Self {
+        let transport_slot = self.transport_slot.clone_into_new_slot();
+        Self { transport_slot }
+    }
+
+    /// Returns whether this sender currently has an available transport.
+    pub(super) fn is_enabled(&self) -> bool {
+        self.transport_slot.is_occupied()
+    }
+}
+
+mod slot {
+    use std::sync::{Arc, RwLock};
+    use std::time::Duration;
+
+    use sentry_types::protocol::v7::Envelope;
+
+    use crate::Transport;
+
+    const READ_EXPECT_MSG: &str = "could not acquire transport read lock";
+    const WRITE_EXPECT_MSG: &str = "could not acquire transport write lock";
+
+    /// A transport slot, which may or may not wrap a [`Transport`].
+    ///
+    /// When initially constructed with [`TransportSlot::new`], this type will be wrapping this
+    /// transport. As long as constructed with a [`Transport`], as intended, this type also
+    /// implements [`Transport`], and all of the method calls forward to the underlying transport.
+    /// However, after [`Transport::shutdown`] is called on this slot, the slot is emptied, and
+    /// all future [`Transport`] method calls become no-ops. The type provides
+    /// [`Self::is_occupied`] to check if the transport is still present.
+    ///
+    /// This type has [`Arc`]-like clone semantics: clones share the underlying transport, and also
+    /// share the slot occupied status.
+    #[derive(Debug)]
+    pub(super) struct TransportSlot<T: ?Sized> {
+        inner: Arc<RwLock<Option<Arc<T>>>>,
+    }
+
+    impl<T: ?Sized> TransportSlot<T> {
+        /// Create a new, occupied [`TransportSlot`] wrapping the provided transport.
+        pub(super) fn new(transport: Arc<T>) -> Self {
+            let inner = Arc::new(RwLock::new(Some(transport)));
+
+            Self { inner }
+        }
+
+        /// Determine whether the slot is occupied, i.e. whether there is a transport inside.
+        pub(super) fn is_occupied(&self) -> bool {
+            self.inner.read().expect(READ_EXPECT_MSG).is_some()
+        }
+
+        /// Creates a new [`TransportSlot`] with the same underlying `Transport`, but in a new
+        /// slot.
+        ///
+        /// If there is no transport, then we just return a clone of this empty slot. As empty
+        /// slots cannot become occupied later, this has the same semantics as returning a new
+        /// empty slot.
+        pub(super) fn clone_into_new_slot(&self) -> Self {
+            self.inner
+                .read()
+                .expect(READ_EXPECT_MSG)
+                .as_ref()
+                .map(|transport| Self::new(transport.clone()))
+                .unwrap_or_else(|| self.clone())
+        }
+    }
+
+    impl<T> TransportSlot<T>
+    where
+        T: Transport + ?Sized,
+    {
+        pub(super) fn send_envelope_with<F>(&self, builder: F)
+        where
+            F: FnOnce() -> Option<Envelope>,
+        {
+            if let Some((transport, envelope)) = self
+                .inner
+                .read()
+                .expect(READ_EXPECT_MSG)
+                .as_deref()
+                .and_then(|transport| Some((transport, builder()?)))
+            {
+                transport.send_envelope(envelope);
+            }
+        }
+
+        pub(super) fn flush(&self, timeout: Duration) -> bool {
+            self.inner
+                .read()
+                .expect(READ_EXPECT_MSG)
+                .as_deref()
+                .map(|transport| transport.flush(timeout))
+                .unwrap_or(true)
+        }
+
+        pub(super) fn shutdown(&self, timeout: Duration) -> bool {
+            let transport_opt = self.inner.write().expect(WRITE_EXPECT_MSG).take();
+            if let Some(transport) = transport_opt {
+                sentry_debug!("client close; request transport to shut down");
+                transport.shutdown(timeout)
+            } else {
+                sentry_debug!("client close; no transport to shut down");
+                true
+            }
+        }
+    }
+
+    impl<T: ?Sized> Clone for TransportSlot<T> {
+        fn clone(&self) -> Self {
+            Self {
+                inner: self.inner.clone(),
+            }
+        }
+    }
+
+    impl<T: ?Sized> Default for TransportSlot<T> {
+        /// Creates an empty, no-op [`TransportSlot`].
+        fn default() -> Self {
+            Self {
+                inner: Default::default(),
+            }
+        }
+    }
+}