Replace cancellation module with stopper

Author: rusty

Cargo.toml

@@ -38,7 +38,6 @@ async-session = { version = "3.0", optional = true }
 async-sse = "4.0.1"
 async-std = { version = "1.6.5", features = ["unstable"] }
 async-trait = "0.1.41"
-event-listener = "2.5.1"
 femme = { version = "2.1.1", optional = true }
 futures-util = "0.3.6"
 http-client = { version = "6.1.0", default-features = false }
@@ -49,6 +48,7 @@ pin-project-lite = "0.2.0"
 route-recognizer = "0.2.0"
 serde = "1.0.117"
 serde_json = "1.0.59"
+stopper = "0.2.0"
 
 [dev-dependencies]
 async-std = { version = "1.6.5", features = ["unstable", "attributes"] }

src/cancellation.rs

@@ -76,7 +76,6 @@ mod route;
 mod router;
 mod server;
 
-pub mod cancellation;
 pub mod convert;
 pub mod listener;
 pub mod log;
@@ -99,6 +98,8 @@ pub use server::Server;
 
 pub use http_types::{self as http, Body, Error, Status, StatusCode};
 
+pub use stopper;
+
 /// Create a new Tide server.
 ///
 /// # Examples

src/lib.rs

@@ -1,6 +1,5 @@
 use super::{is_transient_error, ListenInfo};
 
-use crate::cancellation::StopStreamExt;
 use crate::listener::Listener;
 use crate::{log, Server};
 
@@ -99,7 +98,7 @@ where
             .take()
             .expect("`Listener::bind` must be called before `Listener::accept`");
 
-        let mut incoming = listener.incoming().stop_on(server.stop_token.clone());
+        let mut incoming = server.stopper.clone().stop_stream(listener.incoming());
 
         while let Some(stream) = incoming.next().await {
             match stream {

src/listener/tcp_listener.rs

@@ -1,6 +1,5 @@
 use super::{is_transient_error, ListenInfo};
 
-use crate::cancellation::StopStreamExt;
 use crate::listener::Listener;
 use crate::{log, Server};
 
@@ -97,7 +96,7 @@ where
             .take()
             .expect("`Listener::bind` must be called before `Listener::accept`");
 
-        let mut incoming = listener.incoming().stop_on(server.stop_token.clone());
+        let mut incoming = server.stopper.clone().stop_stream(listener.incoming());
 
         while let Some(stream) = incoming.next().await {
             match stream {

src/listener/unix_listener.rs

@@ -3,7 +3,8 @@
 use async_std::io;
 use async_std::sync::Arc;
 
-use crate::cancellation::StopToken;
+use stopper::Stopper;
+
 #[cfg(feature = "cookies")]
 use crate::cookies;
 use crate::listener::{Listener, ToListener};
@@ -39,7 +40,7 @@ pub struct Server<State> {
     /// We don't use a Mutex around the Vec here because adding a middleware during execution should be an error.
     #[allow(clippy::rc_buffer)]
     middleware: Arc<Vec<Arc<dyn Middleware<State>>>>,
-    pub(crate) stop_token: StopToken,
+    pub(crate) stopper: Stopper,
 }
 
 impl Server<()> {
@@ -115,7 +116,7 @@ where
                 Arc::new(log::LogMiddleware::new()),
             ]),
             state,
-            stop_token: StopToken::never(),
+            stopper: Stopper::new(),
         }
     }
 
@@ -289,7 +290,7 @@ where
             router,
             state,
             middleware,
-            stop_token: _,
+            stopper: _,
         } = self.clone();
 
         let method = req.method().to_owned();
@@ -322,26 +323,26 @@ where
         &self.state
     }
 
-    /// Stops the server when given `stop_token` completes.
+    /// Stops the server when given `stopper` stops.
     ///
     /// # Example
     ///
     /// ```rust
-    /// use tide::cancellation::StopSource;
+    /// use tide::stopper::Stopper;
     ///
     /// let mut app = tide::new();
     ///
-    /// let stop_source = StopSource::new();
+    /// let stopper = Stopper::new();
     ///
-    /// app.stop_on(stop_source.token());
+    /// app.with_stopper(stopper.clone());
     ///
     /// // Runs server...
     ///
     /// // When something happens
-    /// drop(stop_source);
+    /// stopper.stop();
     /// ```
-    pub fn stop_on(&mut self, stop_token: StopToken) -> &mut Self {
-        self.stop_token = stop_token;
+    pub fn with_stopper(&mut self, stopper: Stopper) -> &mut Self {
+        self.stopper = stopper;
         self
     }
 }
@@ -358,7 +359,7 @@ impl<State: Clone> Clone for Server<State> {
             router: self.router.clone(),
             state: self.state.clone(),
             middleware: self.middleware.clone(),
-            stop_token: self.stop_token.clone(),
+            stopper: self.stopper.clone(),
         }
     }
 }