refactor stop logic

examples/repl/main.rs

@@ -268,13 +268,15 @@ async fn start(args: Vec<String>) -> Result<(), failure::Error> {
     let context = Context::new("CLI".into(), Path::new(&args[1]).to_path_buf()).await?;
 
     let ctx = context.clone();
-    std::thread::spawn(move || loop {
+    async_std::task::spawn(async move {
-        if ctx.has_next_event() {
+        loop {
-            if let Ok(event) = ctx.get_next_event() {
+            if ctx.has_next_event() {
-                receive_event(event);
+                if let Ok(event) = ctx.get_next_event() {
+                    receive_event(event);
+                }
+            } else {
+                async_std::task::sleep(std::time::Duration::from_millis(50)).await;
             }
-        } else {
-            std::thread::sleep(std::time::Duration::from_millis(50));
         }
     });
 
@@ -316,6 +318,7 @@ async fn start(args: Vec<String>) -> Result<(), failure::Error> {
             }
             Err(ReadlineError::Interrupted) | Err(ReadlineError::Eof) => {
                 println!("Exiting...");
+                context.stop().await;
                 break;
             }
             Err(err) => {
@@ -324,11 +327,10 @@ async fn start(args: Vec<String>) -> Result<(), failure::Error> {
             }
         }
     }
+
     rl.save_history(".dc-history.txt")?;
     println!("history saved");
 
-    context.stop().await;
-
     Ok(())
 }
 

examples/simple.rs

@@ -39,13 +39,15 @@ async fn main() {
     println!("info: {:#?}", info);
 
     let ctx1 = ctx.clone();
-    std::thread::spawn(move || loop {
+    async_std::task::spawn(async move {
-        if ctx1.has_next_event() {
+        loop {
-            if let Ok(event) = ctx1.get_next_event() {
+            if ctx1.has_next_event() {
-                cb(event);
+                if let Ok(event) = ctx1.get_next_event() {
+                    cb(event);
+                }
+            } else {
+                async_std::task::sleep(time::Duration::from_millis(50));
             }
-        } else {
-            std::thread::sleep(time::Duration::from_millis(50));
         }
     });
 

src/context.rs

@@ -128,8 +128,8 @@ impl Context {
     }
 
     pub async fn run(&self) {
-        if self.inner.scheduler.read().await.is_running() {
+        if self.is_running().await {
-            panic!("Already running");
+            return;
         }
 
         let l = &mut *self.inner.scheduler.write().await;
@@ -137,7 +137,7 @@ impl Context {
     }
 
     pub async fn is_running(&self) -> bool {
-        self.inner.scheduler.read().await.is_running()
+        self.inner.is_running().await
     }
 
     pub async fn stop(&self) {
@@ -480,19 +480,21 @@ impl Context {
 }
 
 impl InnerContext {
-    async fn stop(&self) {
+    async fn is_running(&self) -> bool {
-        if self.scheduler.read().await.is_running() {
+        self.scheduler.read().await.is_running()
-            self.scheduler.write().await.stop().await;
-        }
     }
-}
 
-impl Drop for InnerContext {
+    async fn stop(&self) {
-    fn drop(&mut self) {
+        if self.is_running().await {
-        async_std::task::block_on(async move {
+            let token = {
-            self.stop().await;
+                let lock = &*self.scheduler.read().await;
-            self.sql.close().await;
+                lock.pre_stop().await
-        });
+            };
+            {
+                let lock = &mut *self.scheduler.write().await;
+                lock.stop(token).await;
+            }
+        }
     }
 }
 

src/scheduler.rs

@@ -9,6 +9,8 @@ use crate::imap::Imap;
 use crate::job::{self, Thread};
 use crate::smtp::Smtp;
 
+pub(crate) struct StopToken;
+
 /// Job and connection scheduler.
 #[derive(Debug)]
 pub(crate) enum Scheduler {
@@ -287,8 +289,8 @@ impl Scheduler {
         }
     }
 
-    /// Halt the scheduler, panics if it is already stopped.
+    /// Halts the scheduler, must be called first, and then `stop`.
-    pub async fn stop(&mut self) {
+    pub(crate) async fn pre_stop(&self) -> StopToken {
         match self {
             Scheduler::Stopped => {
                 panic!("WARN: already stopped");
@@ -306,6 +308,19 @@ impl Scheduler {
                     .join(sentbox.stop())
                     .join(smtp.stop())
                     .await;
+
+                StopToken
+            }
+        }
+    }
+
+    /// Halt the scheduler, must only be called after pre_stop.
+    pub(crate) async fn stop(&mut self, _t: StopToken) {
+        match self {
+            Scheduler::Stopped => {
+                panic!("WARN: already stopped");
+            }
+            Scheduler::Running { .. } => {
                 *self = Scheduler::Stopped;
             }
         }