feat(inserter): add callback function support (#307)

Co-authored-by: Serge Klochkov <3175289+slvrtrn@users.noreply.github.com>
Co-authored-by: Austin Bonander <austin.bonander@gmail.com>

Author: 3 people

src/inserter.rs

@@ -32,6 +32,9 @@ pub struct Inserter<T> {
     ticks: Ticks,
     pending: Quantities,
     in_transaction: bool,
+
+    #[allow(clippy::type_complexity)]
+    on_commit: Option<Box<dyn FnMut(&Quantities) + Send + 'static>>,
 }
 
 /// Statistics about pending or inserted data.
@@ -70,6 +73,7 @@ where
             ticks: Ticks::default(),
             pending: Quantities::ZERO,
             in_transaction: false,
+            on_commit: None,
         }
     }
 
@@ -199,6 +203,19 @@ where
         self.ticks.reschedule();
     }
 
+    /// Registers a callback that will be invoked after each successful batch commit.
+    ///
+    /// The callback receives the committed [`Quantities`]. It is invoked only
+    /// when a batch actually commits (i.e., non-zero transactions), and only
+    /// after the commit completes successfully.
+    pub fn with_commit_callback(
+        mut self,
+        callback: impl FnMut(&Quantities) + Send + 'static,
+    ) -> Self {
+        self.on_commit = Some(Box::new(callback));
+        self
+    }
+
     /// How much time we have until the next tick.
     ///
     /// `None` if the period isn't configured.
@@ -258,21 +275,16 @@ where
 
     /// Ends the current `INSERT` unconditionally.
     pub async fn force_commit(&mut self) -> Result<Quantities> {
-        self.in_transaction = false;
-
-        let quantities = mem::replace(&mut self.pending, Quantities::ZERO);
-        let result = self.insert().await;
+        let quantities = self.insert().await?;
         self.ticks.reschedule();
-        result?;
         Ok(quantities)
     }
 
     /// Ends the current `INSERT` and whole `Inserter` unconditionally.
     ///
     /// If it isn't called, the current `INSERT` is aborted.
     pub async fn end(mut self) -> Result<Quantities> {
-        self.insert().await?;
-        Ok(self.pending)
+        self.insert().await
     }
 
     fn limits_reached(&self) -> bool {
@@ -281,11 +293,21 @@ where
             || self.ticks.reached()
     }
 
-    async fn insert(&mut self) -> Result<()> {
+    async fn insert(&mut self) -> Result<Quantities> {
+        self.in_transaction = false;
+        let quantities = mem::replace(&mut self.pending, Quantities::ZERO);
+
         if let Some(insert) = self.insert.take() {
             insert.end().await?;
         }
-        Ok(())
+
+        if let Some(cb) = &mut self.on_commit
+            && quantities.transactions > 0
+        {
+            (cb)(&quantities);
+        }
+
+        Ok(quantities)
     }
 
     #[cold]