Make Imap.fetch() async

src/imap/mod.rs

@@ -430,8 +430,7 @@ impl Imap {
         });
     }
 
-    pub fn fetch(&self, context: &Context, watch_folder: &str) -> Result<()> {
+    pub async fn fetch(&self, context: &Context, watch_folder: &str) -> Result<()> {
-        task::block_on(async move {
             if !context.sql.is_open() {
                 // probably shutdown
                 return Err(Error::InTeardown);
@@ -442,7 +441,6 @@ impl Imap {
                 // We fetch until no more new messages are there.
             }
             Ok(())
-        })
     }
 
     fn get_config_last_seen_uid<S: AsRef<str>>(&self, context: &Context, folder: S) -> (u32, u32) {

src/job.rs

@@ -8,6 +8,8 @@ use std::time::Duration;
 use deltachat_derive::{FromSql, ToSql};
 use rand::{thread_rng, Rng};
 
+use async_std::task;
+
 use crate::blob::BlobObject;
 use crate::chat;
 use crate::config::Config;
@@ -398,31 +400,37 @@ pub fn job_kill_action(context: &Context, action: Action) -> bool {
 pub fn perform_inbox_fetch(context: &Context) {
     let use_network = context.get_config_bool(Config::InboxWatch);
 
+    task::block_on(
     context
         .inbox_thread
         .write()
         .unwrap()
-        .fetch(context, use_network);
+        .fetch(context, use_network),
+    );
 }
 
 pub fn perform_mvbox_fetch(context: &Context) {
     let use_network = context.get_config_bool(Config::MvboxWatch);
 
+    task::block_on(
     context
         .mvbox_thread
         .write()
         .unwrap()
-        .fetch(context, use_network);
+        .fetch(context, use_network),
+    );
 }
 
 pub fn perform_sentbox_fetch(context: &Context) {
     let use_network = context.get_config_bool(Config::SentboxWatch);
 
+    task::block_on(
     context
         .sentbox_thread
         .write()
         .unwrap()
-        .fetch(context, use_network);
+        .fetch(context, use_network),
+    );
 }
 
 pub fn perform_inbox_idle(context: &Context) {

src/job_thread.rs

@@ -73,7 +73,7 @@ impl JobThread {
         info!(context, "Interrupting {}-IDLE... finished", self.name);
     }
 
-    pub fn fetch(&mut self, context: &Context, use_network: bool) {
+    pub async fn fetch(&mut self, context: &Context, use_network: bool) {
         {
             let &(ref lock, _) = &*self.state.clone();
             let mut state = lock.lock().unwrap();
@@ -86,10 +86,10 @@ impl JobThread {
         }
 
         if use_network {
-            if let Err(err) = self.connect_and_fetch(context) {
+            if let Err(err) = self.connect_and_fetch(context).await {
                 warn!(context, "connect+fetch failed: {}, reconnect & retry", err);
                 self.imap.trigger_reconnect();
-                if let Err(err) = self.connect_and_fetch(context) {
+                if let Err(err) = self.connect_and_fetch(context).await {
                     warn!(context, "connect+fetch failed: {}", err);
                 }
             }
@@ -97,14 +97,18 @@ impl JobThread {
         self.state.0.lock().unwrap().using_handle = false;
     }
 
-    fn connect_and_fetch(&mut self, context: &Context) -> Result<()> {
+    async fn connect_and_fetch(&mut self, context: &Context) -> Result<()> {
         let prefix = format!("{}-fetch", self.name);
         match self.imap.connect_configured(context) {
             Ok(()) => {
                 if let Some(watch_folder) = self.get_watch_folder(context) {
                     let start = std::time::Instant::now();
                     info!(context, "{} started...", prefix);
-                    let res = self.imap.fetch(context, &watch_folder).map_err(Into::into);
+                    let res = self
+                        .imap
+                        .fetch(context, &watch_folder)
+                        .await
+                        .map_err(Into::into);
                     let elapsed = start.elapsed().as_millis();
                     info!(context, "{} done in {:.3} ms.", prefix, elapsed);