diff --git a/crates/imap/src/op/idle.rs b/crates/imap/src/op/idle.rs index ff29d1ad..183d5e42 100644 --- a/crates/imap/src/op/idle.rs +++ b/crates/imap/src/op/idle.rs @@ -62,7 +62,7 @@ impl Session { // Register with state manager let mut change_rx = if let Some(change_rx) = self .jmap - .subscribe_state_manager(data.account_id, data.account_id, types) + .subscribe_state_manager(data.account_id, types) .await { change_rx diff --git a/crates/jmap/src/api/event_source.rs b/crates/jmap/src/api/event_source.rs index a49261f4..929abb74 100644 --- a/crates/jmap/src/api/event_source.rs +++ b/crates/jmap/src/api/event_source.rs @@ -110,7 +110,7 @@ impl JMAP { // Register with state manager let mut change_rx = if let Some(change_rx) = self - .subscribe_state_manager(access_token.primary_id(), access_token.primary_id(), types) + .subscribe_state_manager(access_token.primary_id(), types) .await { change_rx diff --git a/crates/jmap/src/services/state.rs b/crates/jmap/src/services/state.rs index e51b0c0a..698c2320 100644 --- a/crates/jmap/src/services/state.rs +++ b/crates/jmap/src/services/state.rs @@ -41,7 +41,6 @@ use super::IPC_CHANNEL_BUFFER; #[derive(Debug)] pub enum Event { Subscribe { - id: u32, account_id: u32, types: Bitmap, tx: mpsc::Sender, @@ -80,13 +79,19 @@ impl Subscriber { } } -const PURGE_EVERY_SECS: u64 = 3600; -const SEND_TIMEOUT_MS: u64 = 500; +const PURGE_EVERY: Duration = Duration::from_secs(3600); +const SEND_TIMEOUT: Duration = Duration::from_millis(500); pub fn init_state_manager() -> (mpsc::Sender, mpsc::Receiver) { mpsc::channel::(IPC_CHANNEL_BUFFER) } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +enum SubscriberId { + Ipc(u32), + Push(u32), +} + #[allow(clippy::unwrap_or_default)] pub fn spawn_state_manager( core: Arc, @@ -96,7 +101,8 @@ pub fn spawn_state_manager( let push_tx = spawn_push_manager(settings); tokio::spawn(async move { - let mut subscribers: AHashMap> = AHashMap::default(); + let mut subscribers: AHashMap> = + AHashMap::default(); let mut shared_accounts: AHashMap> = AHashMap::default(); let mut shared_accounts_map: AHashMap>> = AHashMap::default(); @@ -104,7 +110,7 @@ pub fn spawn_state_manager( let mut last_purge = Instant::now(); while let Some(event) = change_rx.recv().await { - let mut purge_needed = last_purge.elapsed() >= Duration::from_secs(PURGE_EVERY_SECS); + let mut purge_needed = last_purge.elapsed() >= PURGE_EVERY; match event { Event::Stop => { @@ -176,7 +182,6 @@ pub fn spawn_state_manager( shared_accounts.insert(account_id, shared_account_ids); } Event::Subscribe { - id, account_id, types, tx, @@ -185,7 +190,7 @@ pub fn spawn_state_manager( .entry(account_id) .or_insert_with(AHashMap::default) .insert( - u32::MAX - id, + SubscriberId::Ipc(rand::random()), Subscriber { types, subscription: SubscriberType::Ipc { tx }, @@ -226,7 +231,7 @@ pub fn spawn_state_manager( account_id: state_change.account_id, types, }, - Duration::from_millis(SEND_TIMEOUT_MS), + SEND_TIMEOUT, ) .await { @@ -242,7 +247,7 @@ pub fn spawn_state_manager( { push_ids.push(Id::from_parts( *owner_account_id, - *subscriber_id, + (*subscriber_id).into(), )); } _ => { @@ -278,22 +283,20 @@ pub fn spawn_state_manager( let mut remove_ids = Vec::new(); for subscriber_id in subscribers.keys() { - #[allow(clippy::match_like_matches_macro)] - if (*subscriber_id < u32::MAX / 2) - && !subscriptions.iter().any(|s| match s { - UpdateSubscription::Verified( - crate::push::PushSubscription { id, .. }, - ) if id == subscriber_id => true, - _ => false, - }) - { - remove_ids.push(*subscriber_id); + if let SubscriberId::Push(push_id) = subscriber_id { + if !subscriptions.iter().any(|s| { + matches!(s, UpdateSubscription::Verified( + crate::push::PushSubscription { id, .. } + ) if id == push_id) + }) { + remove_ids.push(*subscriber_id); + } } } for remove_id in remove_ids { push_updates.push(crate::push::PushUpdate::Unregister { - id: Id::from_parts(account_id, remove_id), + id: Id::from_parts(account_id, remove_id.into()), }); subscribers.remove(&remove_id); } @@ -321,7 +324,7 @@ pub fn spawn_state_manager( .entry(account_id) .or_insert_with(AHashMap::default) .insert( - verified.id, + SubscriberId::Push(verified.id), Subscriber { types: verified.types, subscription: SubscriberType::Push { @@ -390,7 +393,6 @@ pub fn spawn_state_manager( impl JMAP { pub async fn subscribe_state_manager( &self, - id: u32, account_id: u32, types: Bitmap, ) -> Option> { @@ -400,7 +402,6 @@ impl JMAP { for event in [ Event::UpdateSharedAccounts { account_id }, Event::Subscribe { - id, account_id, types, tx: change_tx, @@ -456,3 +457,12 @@ impl JMAP { true } } + +impl From for u32 { + fn from(subscriber_id: SubscriberId) -> u32 { + match subscriber_id { + SubscriberId::Ipc(id) => id, + SubscriberId::Push(id) => id, + } + } +} diff --git a/crates/jmap/src/websocket/stream.rs b/crates/jmap/src/websocket/stream.rs index e3ed2c23..0915e689 100644 --- a/crates/jmap/src/websocket/stream.rs +++ b/crates/jmap/src/websocket/stream.rs @@ -63,11 +63,7 @@ impl JMAP { // Register with state manager let mut change_rx = if let Some(change_rx) = self - .subscribe_state_manager( - access_token.primary_id(), - access_token.primary_id(), - Bitmap::all(), - ) + .subscribe_state_manager(access_token.primary_id(), Bitmap::all()) .await { change_rx diff --git a/tests/src/imap/idle.rs b/tests/src/imap/idle.rs index d141769b..fa31563c 100644 --- a/tests/src/imap/idle.rs +++ b/tests/src/imap/idle.rs @@ -23,6 +23,8 @@ use imap_proto::ResponseType; +use crate::jmap::delivery::SmtpConnection; + use super::{AssertResult, ImapConnection, Type}; pub async fn test(imap: &mut ImapConnection, imap_check: &mut ImapConnection) { @@ -145,6 +147,28 @@ pub async fn test(imap: &mut ImapConnection, imap_check: &mut ImapConnection) { .await .assert_contains("* 0 EXISTS"); + // Test SMTP delivery notifications + let mut lmtp = SmtpConnection::connect_port(11201).await; + lmtp.ingest( + "bill@example.com", + &["jdoe@example.com"], + concat!( + "From: bill@example.com\r\n", + "To: jdoe@example.com\r\n", + "Subject: TPS Report\r\n", + "X-Spam-Status: No\r\n", + "\r\n", + "I'm going to need those TPS reports ASAP. ", + "So, if you could do that, that'd be great." + ), + ) + .await; + imap_check + .assert_read(Type::Status, ResponseType::Ok) + .await + .assert_contains("STATUS \"INBOX\"") + .assert_contains("MESSAGES 11"); + // Stop IDLE mode imap_check.send_raw("DONE").await; imap_check.assert_read(Type::Tagged, ResponseType::Ok).await; diff --git a/tests/src/imap/mod.rs b/tests/src/imap/mod.rs index 9389798a..11cb18f1 100644 --- a/tests/src/imap/mod.rs +++ b/tests/src/imap/mod.rs @@ -508,7 +508,7 @@ impl ImapConnection { Type::Untagged | Type::Status => "* ", Type::Continuation => "+ ", }); - //println!("<- {:?}", line); + //let c = println!("<- {:?}", line); lines.push(line); if is_done { return lines; @@ -526,20 +526,20 @@ impl ImapConnection { } pub async fn send(&mut self, text: &str) { - //println!("-> {}{:?}", std::str::from_utf8(self.tag).unwrap(), text); + //let c = println!("-> {}{:?}", std::str::from_utf8(self.tag).unwrap(), text); self.writer.write_all(self.tag).await.unwrap(); self.writer.write_all(text.as_bytes()).await.unwrap(); self.writer.write_all(b"\r\n").await.unwrap(); } pub async fn send_untagged(&mut self, text: &str) { - //println!("-> {:?}", text); + //let c = println!("-> {:?}", text); self.writer.write_all(text.as_bytes()).await.unwrap(); self.writer.write_all(b"\r\n").await.unwrap(); } pub async fn send_raw(&mut self, text: &str) { - //println!("-> {:?}", text); + //let c = println!("-> {:?}", text); self.writer.write_all(text.as_bytes()).await.unwrap(); } }