llange
/
meli
forked from meli/meli
1
Fork 0
Code Issues Pull Requests 1 Projects Releases Wiki Activity

melib/imap: don't fetch RFC822 except when requested 

In some cases when handling new server events, the entire body message
was unnecessarily fetched.

Closes #87 IMAP: don't fetch RFC822 except when requested
jmap-eventsource
Manos Pitsidianakis 2020-11-29 15:34:30 +02:00
parent 66dea9148b
commit ddfadc748d
Signed by: Manos Pitsidianakis
GPG Key ID: 73627C2F690DF710
2 changed files with 301 additions and 214 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

273
melib/src/backends/imap/untagged.rs
View File

@ -28,7 +28,6 @@ use crate::backends::{
RefreshEvent, RefreshEvent,
RefreshEventKind::{self, *}, RefreshEventKind::{self, *},
}; };
use crate::email::Envelope;
use crate::error::*; use crate::error::*;
use std::convert::TryInto; use std::convert::TryInto;
@ -196,29 +195,71 @@ impl ImapConnection {
debug!("exists {}", n); debug!("exists {}", n);
try_fail!( try_fail!(
mailbox_hash, mailbox_hash,
self.send_command(format!("FETCH {} (UID FLAGS RFC822)", n).as_bytes()).await self.send_command(format!("FETCH {} (UID FLAGS ENVELOPE BODY.PEEK[HEADER.FIELDS (REFERENCES)] BODYSTRUCTURE)", n).as_bytes()).await
self.read_response(&mut response, RequiredResponses::FETCH_REQUIRED).await self.read_response(&mut response, RequiredResponses::FETCH_REQUIRED).await
); );
match super::protocol_parser::fetch_responses(&response) { let mut v = match super::protocol_parser::fetch_responses(&response) {
Ok((_, v, _)) => { Ok((_, v, _)) => v,
'fetch_responses: for FetchResponse { Err(err) => {
uid, flags, body, .. debug!(
} in v "Error when parsing FETCH response after untagged exists {:?}",
err
);
return Ok(true);
}
};
debug!("responses len is {}", v.len());
for FetchResponse {
ref uid,
ref mut envelope,
ref mut flags,
ref references,
..
} in &mut v
{ {
if uid.is_none() || flags.is_none() || body.is_none() { if uid.is_none() || flags.is_none() || envelope.is_none() {
continue; continue;
} }
let uid = uid.unwrap(); let uid = uid.unwrap();
if self let env = envelope.as_mut().unwrap();
env.set_hash(generate_envelope_hash(&mailbox.imap_path(), &uid));
if let Some(value) = references {
let parse_result = crate::email::parser::address::msg_id_list(value);
if let Ok((_, value)) = parse_result {
let prev_val = env.references.take();
for v in value {
env.push_references(v);
}
if let Some(prev) = prev_val {
for v in prev.refs {
env.push_references(v);
}
}
}
env.set_references(value);
}
let mut tag_lck = self.uid_store.tag_index.write().unwrap();
if let Some((flags, keywords)) = flags {
env.set_flags(*flags);
if !env.is_seen() {
mailbox.unseen.lock().unwrap().insert_new(env.hash());
}
for f in keywords {
let hash = tag_hash!(f);
if !tag_lck.contains_key(&hash) {
tag_lck.insert(hash, f.to_string());
}
env.labels_mut().push(hash);
}
}
mailbox.exists.lock().unwrap().insert_new(env.hash());
if !self
.uid_store .uid_store
.uid_index .uid_index
.lock() .lock()
.unwrap() .unwrap()
.contains_key(&(mailbox_hash, uid)) .contains_key(&(mailbox_hash, uid))
{ {
continue 'fetch_responses;
}
let env_hash = generate_envelope_hash(&mailbox.imap_path(), &uid);
self.uid_store self.uid_store
.msn_index .msn_index
.lock() .lock()
@ -226,64 +267,48 @@ impl ImapConnection {
.entry(mailbox_hash) .entry(mailbox_hash)
.or_default() .or_default()
.push(uid); .push(uid);
if let Ok(mut env) = }
Envelope::from_bytes(body.unwrap(), flags.as_ref().map(|&(f, _)| f))
{
env.set_hash(env_hash);
self.uid_store self.uid_store
.hash_index .hash_index
.lock() .lock()
.unwrap() .unwrap()
.insert(env_hash, (uid, mailbox_hash)); .insert(env.hash(), (uid, mailbox_hash));
self.uid_store self.uid_store
.uid_index .uid_index
.lock() .lock()
.unwrap() .unwrap()
.insert((mailbox_hash, uid), env_hash); .insert((mailbox_hash, uid), env.hash());
if let Some((_, keywords)) = flags {
let mut tag_lck = self.uid_store.tag_index.write().unwrap();
for f in keywords {
let hash = tag_hash!(f);
if !tag_lck.contains_key(&hash) {
tag_lck.insert(hash, f);
}
env.labels_mut().push(hash);
}
}
debug!( debug!(
"Create event {} {} {}", "Create event {} {} {}",
env.hash(), env.hash(),
env.subject(), env.subject(),
mailbox.path(), mailbox.path(),
); );
if !env.is_seen() {
mailbox.unseen.lock().unwrap().insert_new(env.hash());
} }
mailbox.exists.lock().unwrap().insert_new(env.hash());
let mut event: [(UID, RefreshEvent); 1] = [(
uid,
RefreshEvent {
account_hash: self.uid_store.account_hash,
mailbox_hash,
kind: Create(Box::new(env)),
},
)];
if self.uid_store.keep_offline_cache { if self.uid_store.keep_offline_cache {
cache_handle.update(mailbox_hash, &event)?; if let Err(err) = cache_handle
.insert_envelopes(mailbox_hash, &v)
.chain_err_summary(|| {
format!(
"Could not save envelopes in cache for mailbox {}",
&mailbox.imap_path()
)
})
{
crate::log(err.to_string(), crate::INFO);
} }
self.add_refresh_event(std::mem::replace( }
&mut event[0].1, for response in v {
RefreshEvent { if let FetchResponse {
envelope: Some(envelope),
..
} = response
{
self.add_refresh_event(RefreshEvent {
account_hash: self.uid_store.account_hash, account_hash: self.uid_store.account_hash,
mailbox_hash, mailbox_hash,
kind: Rescan, kind: Create(Box::new(envelope)),
}, });
));
}
}
}
Err(e) => {
debug!(e);
} }
} }
} }
@ -308,24 +333,90 @@ impl ImapConnection {
for ms in iter { for ms in iter {
accum = format!("{},{}", accum, to_str!(ms).trim()); accum = format!("{},{}", accum, to_str!(ms).trim());
} }
format!("UID FETCH {} (FLAGS RFC822)", accum) format!("UID FETCH {} (UID FLAGS ENVELOPE BODY.PEEK[HEADER.FIELDS (REFERENCES)] BODYSTRUCTURE)", accum)
}; };
try_fail!( try_fail!(
mailbox_hash, mailbox_hash,
self.send_command(command.as_bytes()).await self.send_command(command.as_bytes()).await
self.read_response(&mut response, RequiredResponses::FETCH_REQUIRED).await self.read_response(&mut response, RequiredResponses::FETCH_REQUIRED).await
); );
debug!(&response); let mut v = match super::protocol_parser::fetch_responses(&response) {
match super::protocol_parser::fetch_responses(&response) { Ok((_, v, _)) => v,
Ok((_, v, _)) => { Err(err) => {
debug!(
"Error when parsing FETCH response after untagged recent {:?}",
err
);
return Ok(true);
}
};
debug!("responses len is {}", v.len());
for FetchResponse { for FetchResponse {
uid, flags, body, .. ref uid,
} in v ref mut envelope,
ref mut flags,
ref references,
..
} in &mut v
{ {
if uid.is_none() || flags.is_none() || body.is_none() { if uid.is_none() || flags.is_none() || envelope.is_none() {
continue; continue;
} }
let uid = uid.unwrap(); let uid = uid.unwrap();
let env = envelope.as_mut().unwrap();
env.set_hash(generate_envelope_hash(&mailbox.imap_path(), &uid));
if let Some(value) = references {
let parse_result =
crate::email::parser::address::msg_id_list(value);
if let Ok((_, value)) = parse_result {
let prev_val = env.references.take();
for v in value {
env.push_references(v);
}
if let Some(prev) = prev_val {
for v in prev.refs {
env.push_references(v);
}
}
}
env.set_references(value);
}
let mut tag_lck = self.uid_store.tag_index.write().unwrap();
if let Some((flags, keywords)) = flags {
env.set_flags(*flags);
if !env.is_seen() {
mailbox.unseen.lock().unwrap().insert_new(env.hash());
}
for f in keywords {
let hash = tag_hash!(f);
if !tag_lck.contains_key(&hash) {
tag_lck.insert(hash, f.to_string());
}
env.labels_mut().push(hash);
}
}
mailbox.exists.lock().unwrap().insert_new(env.hash());
}
if self.uid_store.keep_offline_cache {
if let Err(err) = cache_handle
.insert_envelopes(mailbox_hash, &v)
.chain_err_summary(|| {
format!(
"Could not save envelopes in cache for mailbox {}",
&mailbox.imap_path()
)
})
{
crate::log(err.to_string(), crate::INFO);
}
}
for response in v {
if let FetchResponse {
envelope: Some(envelope),
uid: Some(uid),
..
} = response
{
if !self if !self
.uid_store .uid_store
.uid_index .uid_index
@ -333,71 +424,35 @@ impl ImapConnection {
.unwrap() .unwrap()
.contains_key(&(mailbox_hash, uid)) .contains_key(&(mailbox_hash, uid))
{ {
if let Ok(mut env) = Envelope::from_bytes( self.uid_store
body.unwrap(), .msn_index
flags.as_ref().map(|&(f, _)| f), .lock()
) { .unwrap()
.entry(mailbox_hash)
.or_default()
.push(uid);
}
self.uid_store self.uid_store
.hash_index .hash_index
.lock() .lock()
.unwrap() .unwrap()
.insert(env.hash(), (uid, mailbox_hash)); .insert(envelope.hash(), (uid, mailbox_hash));
self.uid_store self.uid_store
.uid_index .uid_index
.lock() .lock()
.unwrap() .unwrap()
.insert((mailbox_hash, uid), env.hash()); .insert((mailbox_hash, uid), envelope.hash());
debug!( debug!(
"Create event {} {} {}", "Create event {} {} {}",
env.hash(), envelope.hash(),
env.subject(), envelope.subject(),
mailbox.path(), mailbox.path(),
); );
if let Some((_, keywords)) = flags { self.add_refresh_event(RefreshEvent {
let mut tag_lck =
self.uid_store.tag_index.write().unwrap();
for f in keywords {
let hash = tag_hash!(f);
if !tag_lck.contains_key(&hash) {
tag_lck.insert(hash, f);
}
env.labels_mut().push(hash);
}
}
if !env.is_seen() {
mailbox
.unseen
.lock()
.unwrap()
.insert_new(env.hash());
}
mailbox.exists.lock().unwrap().insert_new(env.hash());
let mut event: [(UID, RefreshEvent); 1] = [(
uid,
RefreshEvent {
account_hash: self.uid_store.account_hash, account_hash: self.uid_store.account_hash,
mailbox_hash, mailbox_hash,
kind: Create(Box::new(env)), kind: Create(Box::new(envelope)),
}, });
)];
if self.uid_store.keep_offline_cache {
cache_handle.update(mailbox_hash, &event)?;
}
self.add_refresh_event(std::mem::replace(
&mut event[0].1,
RefreshEvent {
account_hash: self.uid_store.account_hash,
mailbox_hash,
kind: Rescan,
},
));
}
}
}
}
Err(e) => {
debug!(e);
} }
} }
} }

94
melib/src/backends/imap/watch.rs
View File

@ -267,14 +267,16 @@ pub async fn examine_updates(
cmd.push_str(&n.to_string()); cmd.push_str(&n.to_string());
} }
} }
cmd.push_str(" (UID FLAGS RFC822)"); cmd.push_str(
" (UID FLAGS ENVELOPE BODY.PEEK[HEADER.FIELDS (REFERENCES)] BODYSTRUCTURE)",
);
conn.send_command(cmd.as_bytes()).await?; conn.send_command(cmd.as_bytes()).await?;
conn.read_response(&mut response, RequiredResponses::FETCH_REQUIRED) conn.read_response(&mut response, RequiredResponses::FETCH_REQUIRED)
.await?; .await?;
} else if debug!(select_response.exists > mailbox.exists.lock().unwrap().len()) { } else if debug!(select_response.exists > mailbox.exists.lock().unwrap().len()) {
conn.send_command( conn.send_command(
format!( format!(
"FETCH {}:* (UID FLAGS RFC822)", "FETCH {}:* (UID FLAGS ENVELOPE BODY.PEEK[HEADER.FIELDS (REFERENCES)] BODYSTRUCTURE)",
mailbox.exists.lock().unwrap().len() mailbox.exists.lock().unwrap().len()
) )
.as_bytes(), .as_bytes(),
@ -285,42 +287,70 @@ pub async fn examine_updates(
} else { } else {
return Ok(()); return Ok(());
} }
debug!(&response); debug!(
"fetch response is {} bytes and {} lines",
response.len(),
String::from_utf8_lossy(&response).lines().count()
);
let (_, mut v, _) = protocol_parser::fetch_responses(&response)?; let (_, mut v, _) = protocol_parser::fetch_responses(&response)?;
debug!("responses len is {}", v.len());
for FetchResponse { for FetchResponse {
ref uid, ref uid,
ref mut flags,
ref mut body,
ref mut envelope, ref mut envelope,
ref mut flags,
ref references,
.. ..
} in v.iter_mut() } in v.iter_mut()
{ {
let uid = uid.unwrap(); let uid = uid.unwrap();
*envelope = Envelope::from_bytes(body.take().unwrap(), flags.as_ref().map(|&(f, _)| f)) let env = envelope.as_mut().unwrap();
.map(|mut env| {
env.set_hash(generate_envelope_hash(&mailbox.imap_path(), &uid)); env.set_hash(generate_envelope_hash(&mailbox.imap_path(), &uid));
if let Some((_, keywords)) = flags.take() { if let Some(value) = references {
let parse_result = crate::email::parser::address::msg_id_list(value);
if let Ok((_, value)) = parse_result {
let prev_val = env.references.take();
for v in value {
env.push_references(v);
}
if let Some(prev) = prev_val {
for v in prev.refs {
env.push_references(v);
}
}
}
env.set_references(value);
}
let mut tag_lck = uid_store.tag_index.write().unwrap(); let mut tag_lck = uid_store.tag_index.write().unwrap();
if let Some((flags, keywords)) = flags {
env.set_flags(*flags);
if !env.is_seen() {
mailbox.unseen.lock().unwrap().insert_new(env.hash());
}
mailbox.exists.lock().unwrap().insert_new(env.hash());
for f in keywords { for f in keywords {
let hash = tag_hash!(f); let hash = tag_hash!(f);
if !tag_lck.contains_key(&hash) { if !tag_lck.contains_key(&hash) {
tag_lck.insert(hash, f); tag_lck.insert(hash, f.to_string());
} }
env.labels_mut().push(hash); env.labels_mut().push(hash);
} }
} }
env
})
.map_err(|err| {
debug!("uid {} envelope parse error {}", uid, &err);
err
})
.ok();
} }
if uid_store.keep_offline_cache { if uid_store.keep_offline_cache {
cache_handle.insert_envelopes(mailbox_hash, &v)?; debug!(cache_handle
.insert_envelopes(mailbox_hash, &v)
.chain_err_summary(|| {
format!(
"Could not save envelopes in cache for mailbox {}",
mailbox.imap_path()
)
}))?;
}
for FetchResponse { uid, envelope, .. } in v {
if uid.is_none() || envelope.is_none() {
continue;
} }
'fetch_responses_c: for FetchResponse { uid, envelope, .. } in v {
let uid = uid.unwrap(); let uid = uid.unwrap();
if uid_store if uid_store
.uid_index .uid_index
@ -328,9 +358,22 @@ pub async fn examine_updates(
.unwrap() .unwrap()
.contains_key(&(mailbox_hash, uid)) .contains_key(&(mailbox_hash, uid))
{ {
continue 'fetch_responses_c; continue;
} }
if let Some(env) = envelope { let env = envelope.unwrap();
debug!(
"Create event {} {} {}",
env.hash(),
env.subject(),
mailbox.path(),
);
uid_store
.msn_index
.lock()
.unwrap()
.entry(mailbox_hash)
.or_default()
.push(uid);
uid_store uid_store
.hash_index .hash_index
.lock() .lock()
@ -341,16 +384,6 @@ pub async fn examine_updates(
.lock() .lock()
.unwrap() .unwrap()
.insert((mailbox_hash, uid), env.hash()); .insert((mailbox_hash, uid), env.hash());
debug!(
"Create event {} {} {}",
env.hash(),
env.subject(),
mailbox.path(),
);
if !env.is_seen() {
mailbox.unseen.lock().unwrap().insert_new(env.hash());
}
mailbox.exists.lock().unwrap().insert_new(env.hash());
conn.add_refresh_event(RefreshEvent { conn.add_refresh_event(RefreshEvent {
account_hash: uid_store.account_hash, account_hash: uid_store.account_hash,
mailbox_hash, mailbox_hash,
@ -358,6 +391,5 @@ pub async fn examine_updates(
}); });
} }
} }
}
Ok(()) Ok(())
} }