refactor(persistence): get rid of the StopFuture and return a BoxFutu…

…re instead

The StopFuture does not introduce anything and is not needed. We now directly use a JoinHandle
and return a BoxFuture that awaits for it instead

zebus/src/persistence/mod.rs

@@ -1,6 +1,5 @@
 pub(crate) mod command;
 pub mod event;
-pub mod future;
 mod service;
 pub mod transport;
 

zebus/src/persistence/service.rs

@@ -826,7 +826,7 @@ pub(super) fn spawn<T, S>(
 ) -> (
     mpsc::Sender<PersistenceRequest>,
     BoxStream<'static, PersistenceEvent>,
-    super::future::StopFuture,
+    tokio::task::JoinHandle<Result<(), PersistenceError>>,
 )
 where
     S: Stream<Item = BusEvent> + Send + 'static,
@@ -849,8 +849,9 @@ where
             shutdown,
         };
 
-        let fut = super::future::StopFuture::spawn(service.run(bus_events_rx));
-        (tx, events_rx.boxed(), fut)
+        //let fut = super::future::StopFuture::spawn(service.run(bus_events_rx));
+        let handle = tokio::spawn(service.run(bus_events_rx));
+        (tx, events_rx.boxed(), handle)
     } else {
         let service = TransientService {
             inner,
@@ -859,13 +860,12 @@ where
             shutdown,
         };
 
-        let fut = super::future::StopFuture::spawn(service.run());
-
         let events_rx = futures_util::stream::iter([
             PersistenceEvent::ReplayStarted,
             PersistenceEvent::SafetyStarted,
             PersistenceEvent::Normal,
         ]);
-        (tx, events_rx.boxed(), fut)
+        let handle = tokio::spawn(service.run());
+        (tx, events_rx.boxed(), handle)
     }
 }

zebus/src/persistence/transport.rs

@@ -65,8 +65,8 @@ enum Inner<T> {
         /// Token to cancel task
         shutdown: CancellationToken,
 
-        /// Future that resolves when the task has been joined
-        stop: super::future::StopFuture,
+        /// Join handle for the persistence task
+        handle: tokio::task::JoinHandle<Result<(), PersistenceError>>,
     },
 }
 
@@ -96,7 +96,7 @@ where
     type MessageStream = super::MessageStream;
 
     type StartCompletionFuture = BoxFuture<'static, Result<(), Self::Err>>;
-    type StopCompletionFuture = super::future::StopFuture;
+    type StopCompletionFuture = BoxFuture<'static, Result<(), Self::Err>>;
     type SendFuture = transport::future::SendFuture<PersistenceRequest, Self::Err>;
 
     fn configure(
@@ -174,7 +174,7 @@ where
                 let shutdown = CancellationToken::new();
 
                 // Spawn a new task for the persistence service
-                let (requests_tx, mut events_rx, stop) = super::service::spawn(
+                let (requests_tx, mut events_rx, handle) = super::service::spawn(
                     &configuration,
                     directory,
                     event_rx.stream(),
@@ -210,7 +210,7 @@ where
                         messages_tx,
                         requests_tx,
                         shutdown,
-                        stop,
+                        handle,
                     }),
                     Ok(completion),
                 )
@@ -224,14 +224,21 @@ where
 
     fn stop(&mut self) -> Result<Self::StopCompletionFuture, Self::Err> {
         match self.inner.take() {
-            Some(Inner::Started { shutdown, stop, .. }) => {
+            Some(Inner::Started {
+                shutdown, handle, ..
+            }) => {
                 // Cancel the task
                 shutdown.cancel();
 
                 // TODO(oktal): we should go back to the `Configured` state but we need to retrieve
                 // the inner transport layer that was moved to an inner task
 
-                Ok(stop)
+                Ok(Box::pin(async move {
+                    match handle.await {
+                        Ok(res) => res,
+                        Err(e) => Err(PersistenceError::Join(e)),
+                    }
+                }))
             }
             _ => Err(PersistenceError::InvalidOperation),
         }