Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-05-08 01:16:31 +03:00
Code Issues Packages Projects Releases Wiki Activity

cleanup select_folder and fix idle/termination issues (#822)

Browse Source 
cleanup select_folder and fix idle/termination issues
This commit is contained in:
Friedel Ziegelmayer
2019-11-11 16:47:06 +01:00
committed by GitHub
parent 01fe782fa0 3dc589788c
commit b5075a7122
2 changed files with 103 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
python/tests/test_account.py
View File

@@ -675,7 +675,6 @@ class TestOnlineAccount:
assert len(messages) == 1 assert len(messages) == 1
assert messages[0].text == "msg1" assert messages[0].text == "msg1"
pytest.xfail("cannot export twice yet, probably due to interrupt_idle failing")
# wait until a second passed since last backup # wait until a second passed since last backup
# because get_latest_backupfile() shall return the latest backup # because get_latest_backupfile() shall return the latest backup
# from a UI it's unlikely anyone manages to export two # from a UI it's unlikely anyone manages to export two
@@ -896,7 +895,7 @@ class TestOnlineConfigureFails:
ac1.start_threads() ac1.start_threads()
wait_configuration_progress(ac1, 500) wait_configuration_progress(ac1, 500)
ev1 = ac1._evlogger.get_matching("DC_EVENT_ERROR_NETWORK") ev1 = ac1._evlogger.get_matching("DC_EVENT_ERROR_NETWORK")
assert "authentication failed" in ev1[2].lower() assert "cannot login" in ev1[2].lower()
wait_configuration_progress(ac1, 0, 0) wait_configuration_progress(ac1, 0, 0)
def test_invalid_user(self, acfactory): def test_invalid_user(self, acfactory):
@@ -905,7 +904,7 @@ class TestOnlineConfigureFails:
ac1.start_threads() ac1.start_threads()
wait_configuration_progress(ac1, 500) wait_configuration_progress(ac1, 500)
ev1 = ac1._evlogger.get_matching("DC_EVENT_ERROR_NETWORK") ev1 = ac1._evlogger.get_matching("DC_EVENT_ERROR_NETWORK")
assert "authentication failed" in ev1[2].lower() assert "cannot login" in ev1[2].lower()
wait_configuration_progress(ac1, 0, 0) wait_configuration_progress(ac1, 0, 0)
def test_invalid_domain(self, acfactory): def test_invalid_domain(self, acfactory):

112
src/imap.rs
View File

@@ -45,6 +45,7 @@ pub struct Imap {
session: Arc<Mutex<Option<Session>>>, session: Arc<Mutex<Option<Session>>>,
connected: Arc<Mutex<bool>>, connected: Arc<Mutex<bool>>,
interrupt: Arc<Mutex<Option<stop_token::StopSource>>>, interrupt: Arc<Mutex<Option<stop_token::StopSource>>>,
skip_next_idle_wait: Arc<Mutex<bool>>,
should_reconnect: AtomicBool, should_reconnect: AtomicBool,
} }
@@ -119,6 +120,7 @@ impl Imap {
config: Arc::new(RwLock::new(ImapConfig::default())), config: Arc::new(RwLock::new(ImapConfig::default())),
interrupt: Arc::new(Mutex::new(None)), interrupt: Arc::new(Mutex::new(None)),
connected: Arc::new(Mutex::new(false)), connected: Arc::new(Mutex::new(false)),
skip_next_idle_wait: Arc::new(Mutex::new(false)),
should_reconnect: AtomicBool::new(false), should_reconnect: AtomicBool::new(false),
} }
} }
@@ -396,12 +398,16 @@ impl Imap {
}) })
} }
async fn select_folder<S: AsRef<str>>(&self, context: &Context, folder: Option<S>) -> usize { async fn select_folder<S: AsRef<str>>(
&self,
context: &Context,
folder: Option<S>,
) -> ImapActionResult {
if self.session.lock().await.is_none() { if self.session.lock().await.is_none() {
let mut cfg = self.config.write().await; let mut cfg = self.config.write().await;
cfg.selected_folder = None; cfg.selected_folder = None;
cfg.selected_folder_needs_expunge = false; cfg.selected_folder_needs_expunge = false;
return 0; return ImapActionResult::Failed;
} }
// if there is a new folder and the new folder is equal to the selected one, there's nothing to do. // if there is a new folder and the new folder is equal to the selected one, there's nothing to do.
@@ -409,7 +415,7 @@ impl Imap {
if let Some(ref folder) = folder { if let Some(ref folder) = folder {
if let Some(ref selected_folder) = self.config.read().await.selected_folder { if let Some(ref selected_folder) = self.config.read().await.selected_folder {
if folder.as_ref() == selected_folder { if folder.as_ref() == selected_folder {
return 1; return ImapActionResult::AlreadyDone;
} }
} }
} }
@@ -429,11 +435,11 @@ impl Imap {
} }
Err(err) => { Err(err) => {
warn!(context, "failed to close session: {:?}", err); warn!(context, "failed to close session: {:?}", err);
return 0; return ImapActionResult::Failed;
} }
} }
} else { } else {
return 0; return ImapActionResult::Failed;
} }
} }
self.config.write().await.selected_folder_needs_expunge = false; self.config.write().await.selected_folder_needs_expunge = false;
@@ -449,7 +455,7 @@ impl Imap {
config.selected_mailbox = Some(mailbox); config.selected_mailbox = Some(mailbox);
} }
Err(err) => { Err(err) => {
info!( warn!(
context, context,
"Cannot select folder: {}; {:?}.", "Cannot select folder: {}; {:?}.",
folder.as_ref(), folder.as_ref(),
@@ -458,7 +464,7 @@ impl Imap {
self.config.write().await.selected_folder = None; self.config.write().await.selected_folder = None;
self.should_reconnect.store(true, Ordering::Relaxed); self.should_reconnect.store(true, Ordering::Relaxed);
return 0; return ImapActionResult::Failed;
} }
} }
} else { } else {
@@ -466,7 +472,7 @@ impl Imap {
} }
} }
1 ImapActionResult::Success
} }
fn get_config_last_seen_uid<S: AsRef<str>>(&self, context: &Context, folder: S) -> (u32, u32) { fn get_config_last_seen_uid<S: AsRef<str>>(&self, context: &Context, folder: S) -> (u32, u32) {
@@ -492,25 +498,17 @@ impl Imap {
} }
async fn fetch_from_single_folder<S: AsRef<str>>(&self, context: &Context, folder: S) -> usize { async fn fetch_from_single_folder<S: AsRef<str>>(&self, context: &Context, folder: S) -> usize {
if !self.is_connected().await { match self.select_folder(context, Some(&folder)).await {
info!( ImapActionResult::Failed | ImapActionResult::RetryLater => {
context, warn!(
"Cannot fetch from \"{}\" - not connected.",
folder.as_ref()
);
return 0;
}
if self.select_folder(context, Some(&folder)).await == 0 {
info!(
context, context,
"Cannot select folder \"{}\" for fetching.", "Cannot select folder \"{}\" for fetching.",
folder.as_ref() folder.as_ref()
); );
return 0; return 0;
} }
ImapActionResult::Success | ImapActionResult::AlreadyDone => {}
}
// compare last seen UIDVALIDITY against the current one // compare last seen UIDVALIDITY against the current one
let (mut uid_validity, mut last_seen_uid) = self.get_config_last_seen_uid(context, &folder); let (mut uid_validity, mut last_seen_uid) = self.get_config_last_seen_uid(context, &folder);
@@ -748,6 +746,9 @@ impl Imap {
pub fn idle(&self, context: &Context) { pub fn idle(&self, context: &Context) {
task::block_on(async move { task::block_on(async move {
if self.config.read().await.selected_folder.is_none() {
return;
}
if !self.config.read().await.can_idle { if !self.config.read().await.can_idle {
self.fake_idle(context).await; self.fake_idle(context).await;
return; return;
@@ -756,12 +757,19 @@ impl Imap {
self.setup_handle_if_needed(context); self.setup_handle_if_needed(context);
let watch_folder = self.config.read().await.watch_folder.clone(); let watch_folder = self.config.read().await.watch_folder.clone();
if self.select_folder(context, watch_folder.as_ref()).await == 0 { match self.select_folder(context, watch_folder.as_ref()).await {
warn!(context, "IMAP-IDLE not setup."); ImapActionResult::Success | ImapActionResult::AlreadyDone => {}
ImapActionResult::Failed | ImapActionResult::RetryLater => {
warn!(
context,
"idle select_folder failed {:?}",
watch_folder.as_ref()
);
self.fake_idle(context).await; self.fake_idle(context).await;
return; return;
} }
}
let session = self.session.lock().await.take(); let session = self.session.lock().await.take();
let timeout = Duration::from_secs(23 * 60); let timeout = Duration::from_secs(23 * 60);
@@ -775,8 +783,17 @@ impl Imap {
let (idle_wait, interrupt) = handle.wait_with_timeout(timeout); let (idle_wait, interrupt) = handle.wait_with_timeout(timeout);
*self.interrupt.lock().await = Some(interrupt); *self.interrupt.lock().await = Some(interrupt);
if *self.skip_next_idle_wait.lock().await {
// interrupt_idle has happened before we
// provided self.interrupt
*self.skip_next_idle_wait.lock().await = false;
std::mem::drop(idle_wait);
info!(context, "Idle wait was skipped");
} else {
info!(context, "Idle entering wait-on-remote state");
let res = idle_wait.await; let res = idle_wait.await;
info!(context, "Idle finished: {:?}", res); info!(context, "Idle finished wait-on-remote: {:?}", res);
}
match handle.done().await { match handle.done().await {
Ok(session) => { Ok(session) => {
*self.session.lock().await = Some(Session::Secure(session)); *self.session.lock().await = Some(Session::Secure(session));
@@ -794,8 +811,18 @@ impl Imap {
let (idle_wait, interrupt) = handle.wait_with_timeout(timeout); let (idle_wait, interrupt) = handle.wait_with_timeout(timeout);
*self.interrupt.lock().await = Some(interrupt); *self.interrupt.lock().await = Some(interrupt);
if *self.skip_next_idle_wait.lock().await {
// interrupt_idle has happened before we
// provided self.interrupt
*self.skip_next_idle_wait.lock().await = false;
std::mem::drop(idle_wait);
info!(context, "Idle wait was skipped");
} else {
info!(context, "Idle entering wait-on-remote state");
let res = idle_wait.await; let res = idle_wait.await;
info!(context, "Idle finished: {:?}", res); info!(context, "Idle finished wait-on-remote: {:?}", res);
}
match handle.done().await { match handle.done().await {
Ok(session) => { Ok(session) => {
*self.session.lock().await = Some(Session::Insecure(session)); *self.session.lock().await = Some(Session::Insecure(session));
@@ -863,7 +890,10 @@ impl Imap {
pub fn interrupt_idle(&self) { pub fn interrupt_idle(&self) {
task::block_on(async move { task::block_on(async move {
let _ = self.interrupt.lock().await.take(); if self.interrupt.lock().await.take().is_none() {
// idle wait is not running
*self.skip_next_idle_wait.lock().await = true;
}
}); });
} }
@@ -999,14 +1029,16 @@ impl Imap {
return Some(ImapActionResult::RetryLater); return Some(ImapActionResult::RetryLater);
} }
} }
if self.select_folder(context, Some(&folder)).await == 0 { match self.select_folder(context, Some(&folder)).await {
ImapActionResult::Success | ImapActionResult::AlreadyDone => None,
res => {
warn!( warn!(
context, context,
"Cannot select folder {} for preparing IMAP operation", folder "Cannot select folder {} for preparing IMAP operation", folder
); );
Some(ImapActionResult::RetryLater)
} else { Some(res)
None }
} }
}) })
} }
@@ -1225,11 +1257,12 @@ impl Imap {
task::block_on(async move { task::block_on(async move {
info!(context, "emptying folder {}", folder); info!(context, "emptying folder {}", folder);
if folder.is_empty() || self.select_folder(context, Some(&folder)).await == 0 { if folder.is_empty() {
warn!(context, "Cannot select folder '{}' for emptying", folder); warn!(context, "cannot perform empty, folder not set");
return; return;
} }
match self.select_folder(context, Some(&folder)).await {
ImapActionResult::Success | ImapActionResult::AlreadyDone => {
if !self if !self
.add_flag_finalized_with_set(context, SELECT_ALL, "\\Deleted") .add_flag_finalized_with_set(context, SELECT_ALL, "\\Deleted")
.await .await
@@ -1238,13 +1271,20 @@ impl Imap {
} else { } else {
// we now trigger expunge to actually delete messages // we now trigger expunge to actually delete messages
self.config.write().await.selected_folder_needs_expunge = true; self.config.write().await.selected_folder_needs_expunge = true;
if self.select_folder::<String>(context, None).await == 0 { if self.select_folder::<String>(context, None).await
== ImapActionResult::Success
{
emit_event!(context, Event::ImapFolderEmptied(folder.to_string()));
} else {
warn!( warn!(
context, context,
"could not perform expunge on empty-marked folder {}", folder "could not perform expunge on empty-marked folder {}", folder
); );
} else { }
emit_event!(context, Event::ImapFolderEmptied(folder.to_string())); }
}
ImapActionResult::Failed | ImapActionResult::RetryLater => {
warn!(context, "could not select folder {}", folder);
} }
} }
}); });