feat: IMAP COMPRESS support

Author: link2xt

Cargo.lock

@@ -41,7 +41,7 @@ ratelimit = { path = "./deltachat-ratelimit" }
 anyhow = { workspace = true }
 async-broadcast = "0.7.1"
 async-channel = { workspace = true }
-async-imap = { version = "0.10.1", default-features = false, features = ["runtime-tokio"] }
+async-imap = { git = "https://github.com/async-email/async-imap.git", default-features = false, features = ["runtime-tokio", "compress"], branch = "link2xt/imap-compress" }
 async-native-tls = { version = "0.5", default-features = false, features = ["runtime-tokio"] }
 async-smtp = { version = "0.9", default-features = false, features = ["runtime-tokio"] }
 async_zip = { version = "0.0.12", default-features = false, features = ["deflate", "fs"] }

Cargo.toml

@@ -39,6 +39,7 @@ use crate::login_param::{
 use crate::message::{self, Message, MessageState, MessengerMessage, MsgId, Viewtype};
 use crate::mimeparser;
 use crate::net::proxy::ProxyConfig;
+use crate::net::session::SessionStream;
 use crate::oauth2::get_oauth2_access_token;
 use crate::receive_imf::{
     from_field_to_contact_id, get_prefetch_parent_message, receive_imf_inner, ReceivedMsg,
@@ -55,7 +56,7 @@ pub mod scan_folders;
 pub mod select_folder;
 pub(crate) mod session;
 
-use client::Client;
+use client::{determine_capabilities, Client};
 use mailparse::SingleInfo;
 use session::Session;
 
@@ -376,7 +377,23 @@ impl Imap {
             };
 
             match login_res {
-                Ok(session) => {
+                Ok(mut session) => {
+                    let capabilities = determine_capabilities(&mut session).await?;
+
+                    let session = if capabilities.can_compress {
+                        info!(context, "Enabling IMAP compression.");
+                        let compressed_session = session
+                            .compress(|s| {
+                                let session_stream: Box<dyn SessionStream> = Box::new(s);
+                                session_stream
+                            })
+                            .await
+                            .context("Failed to enable IMAP compression")?;
+                        Session::new(compressed_session, capabilities)
+                    } else {
+                        Session::new(session, capabilities)
+                    };
+
                     // Store server ID in the context to display in account info.
                     let mut lock = context.server_id.write().await;
                     lock.clone_from(&session.capabilities.server_id);

src/imap.rs

@@ -25,6 +25,10 @@ pub(crate) struct Capabilities {
     /// <https://tools.ietf.org/html/rfc5464>
     pub can_metadata: bool,
 
+    /// True if the server has COMPRESS=DEFLATE capability as defined in
+    /// <https://tools.ietf.org/html/rfc4978>
+    pub can_compress: bool,
+
     /// True if the server supports XDELTAPUSH capability.
     /// This capability means setting /private/devicetoken IMAP METADATA
     /// on the INBOX results in new mail notifications

src/imap/capabilities.rs

@@ -7,7 +7,6 @@ use async_imap::Session as ImapSession;
 use tokio::io::BufWriter;
 
 use super::capabilities::Capabilities;
-use super::session::Session;
 use crate::context::Context;
 use crate::login_param::{ConnectionCandidate, ConnectionSecurity};
 use crate::net::dns::{lookup_host_with_cache, update_connect_timestamp};
@@ -51,7 +50,7 @@ fn alpn(port: u16) -> &'static [&'static str] {
 /// Determine server capabilities.
 ///
 /// If server supports ID capability, send our client ID.
-async fn determine_capabilities(
+pub(crate) async fn determine_capabilities(
     session: &mut ImapSession<Box<dyn SessionStream>>,
 ) -> Result<Capabilities> {
     let caps = session
@@ -69,6 +68,7 @@ async fn determine_capabilities(
         can_check_quota: caps.has_str("QUOTA"),
         can_condstore: caps.has_str("CONDSTORE"),
         can_metadata: caps.has_str("METADATA"),
+        can_compress: caps.has_str("COMPRESS=DEFLATE"),
         can_push: caps.has_str("XDELTAPUSH"),
         is_chatmail: caps.has_str("XCHATMAIL"),
         server_id,
@@ -83,28 +83,31 @@ impl Client {
         }
     }
 
-    pub(crate) async fn login(self, username: &str, password: &str) -> Result<Session> {
+    pub(crate) async fn login(
+        self,
+        username: &str,
+        password: &str,
+    ) -> Result<ImapSession<Box<dyn SessionStream>>> {
         let Client { inner, .. } = self;
-        let mut session = inner
+
+        let session = inner
             .login(username, password)
             .await
             .map_err(|(err, _client)| err)?;
-        let capabilities = determine_capabilities(&mut session).await?;
-        Ok(Session::new(session, capabilities))
+        Ok(session)
     }
 
     pub(crate) async fn authenticate(
         self,
         auth_type: &str,
         authenticator: impl async_imap::Authenticator,
-    ) -> Result<Session> {
+    ) -> Result<ImapSession<Box<dyn SessionStream>>> {
         let Client { inner, .. } = self;
-        let mut session = inner
+        let session = inner
             .authenticate(auth_type, authenticator)
             .await
             .map_err(|(err, _client)| err)?;
-        let capabilities = determine_capabilities(&mut session).await?;
-        Ok(Session::new(session, capabilities))
+        Ok(session)
     }
 
     async fn connection_attempt(

src/imap/client.rs

@@ -53,6 +53,11 @@ impl<T: SessionStream> SessionStream for shadowsocks::ProxyClientStream<T> {
         self.get_mut().set_read_timeout(timeout)
     }
 }
+impl<T: SessionStream> SessionStream for async_imap::DeflateStream<T> {
+    fn set_read_timeout(&mut self, timeout: Option<Duration>) {
+        self.with_lock(|mut stream| stream.set_read_timeout(timeout))
+    }
+}
 
 /// Session stream with a read buffer.
 pub(crate) trait SessionBufStream: SessionStream + AsyncBufRead {}