chore: update to async-channel 2

Cargo.toml

@@ -32,7 +32,7 @@ format-flowed = { path = "./format-flowed" }
 ratelimit = { path = "./deltachat-ratelimit" }
 
 anyhow = "1"
-async-channel = "1.8.0"
+async-channel = "2.0.0"
 async-imap = { version = "0.9.1", default-features = false, features = ["runtime-tokio"] }
 async-native-tls = { version = "0.5", default-features = false, features = ["runtime-tokio"] }
 async-smtp = { version = "0.9", default-features = false, features = ["runtime-tokio"] }
@@ -65,6 +65,7 @@ once_cell = "1.18.0"
 percent-encoding = "2.3"
 parking_lot = "0.12"
 pgp = { version = "0.10", default-features = false }
+pin-project = "1"
 pretty_env_logger = { version = "0.5", optional = true }
 qrcodegen = "1.7.0"
 quick-xml = "0.31"

deltachat-jsonrpc/Cargo.toml

@@ -19,7 +19,7 @@ schemars = "0.8.13"
 serde = { version = "1.0", features = ["derive"] }
 tempfile = "3.8.0"
 log = "0.4"
-async-channel = { version = "1.8.0" }
+async-channel = { version = "2.0.0" }
 futures = { version = "0.3.28" }
 serde_json = "1.0.105"
 yerpc = { version = "0.5.2", features = ["anyhow_expose", "openrpc"] }

deltachat-jsonrpc/src/lib.rs

@@ -17,7 +17,7 @@ mod tests {
         let accounts = Accounts::new(tmp_dir, writable).await?;
         let api = CommandApi::new(accounts);
 
-        let (sender, mut receiver) = unbounded::<String>();
+        let (sender, receiver) = unbounded::<String>();
 
         let (client, mut rx) = RpcClient::new();
         let session = RpcSession::new(client, api);
@@ -36,17 +36,17 @@ mod tests {
             let request = r#"{"jsonrpc":"2.0","method":"add_account","params":[],"id":1}"#;
             let response = r#"{"jsonrpc":"2.0","id":1,"result":1}"#;
             session.handle_incoming(request).await;
-            let result = receiver.next().await;
+            let result = receiver.recv().await?;
             println!("{result:?}");
-            assert_eq!(result, Some(response.to_owned()));
+            assert_eq!(result, response.to_owned());
         }
         {
             let request = r#"{"jsonrpc":"2.0","method":"get_all_account_ids","params":[],"id":2}"#;
             let response = r#"{"jsonrpc":"2.0","id":2,"result":[1]}"#;
             session.handle_incoming(request).await;
-            let result = receiver.next().await;
+            let result = receiver.recv().await?;
             println!("{result:?}");
-            assert_eq!(result, Some(response.to_owned()));
+            assert_eq!(result, response.to_owned());
         }
 
         Ok(())
@@ -59,7 +59,7 @@ mod tests {
         let accounts = Accounts::new(tmp_dir, writable).await?;
         let api = CommandApi::new(accounts);
 
-        let (sender, mut receiver) = unbounded::<String>();
+        let (sender, receiver) = unbounded::<String>();
 
         let (client, mut rx) = RpcClient::new();
         let session = RpcSession::new(client, api);
@@ -78,15 +78,15 @@ mod tests {
             let request = r#"{"jsonrpc":"2.0","method":"add_account","params":[],"id":1}"#;
             let response = r#"{"jsonrpc":"2.0","id":1,"result":1}"#;
             session.handle_incoming(request).await;
-            let result = receiver.next().await;
-            assert_eq!(result, Some(response.to_owned()));
+            let result = receiver.recv().await?;
+            assert_eq!(result, response.to_owned());
         }
         {
             let request = r#"{"jsonrpc":"2.0","method":"batch_set_config","id":2,"params":[1,{"addr":"","mail_user":"","mail_pw":"","mail_server":"","mail_port":"","mail_security":"","imap_certificate_checks":"","send_user":"","send_pw":"","send_server":"","send_port":"","send_security":"","smtp_certificate_checks":"","socks5_enabled":"0","socks5_host":"","socks5_port":"","socks5_user":"","socks5_password":""}]}"#;
             let response = r#"{"jsonrpc":"2.0","id":2,"result":null}"#;
             session.handle_incoming(request).await;
-            let result = receiver.next().await;
-            assert_eq!(result, Some(response.to_owned()));
+            let result = receiver.recv().await?;
+            assert_eq!(result, response.to_owned());
         }
 
         Ok(())

deny.toml

@@ -11,6 +11,7 @@ ignore = [
 # when upgrading.
 # Please keep this list alphabetically sorted.
 skip = [
+     { name = "async-channel", version = "1.9.0" },
      { name = "base16ct", version = "0.1.1" },
      { name = "base64", version = "<0.21" },
      { name = "bitflags", version = "1.3.2" },
@@ -24,6 +25,7 @@ skip = [
      { name = "digest", version = "<0.10" },
      { name = "ed25519-dalek", version = "1.0.1" },
      { name = "ed25519", version = "1.5.3" },
+     { name = "event-listener", version = "2.5.3" },
      { name = "getrandom", version = "<0.2" },
      { name = "hashbrown", version = "<0.14.0" },
      { name = "indexmap", version = "<2.0.0" },

src/events.rs

@@ -1,6 +1,7 @@
 //! # Events specification.
 
 use async_channel::{self as channel, Receiver, Sender, TrySendError};
+use pin_project::pin_project;
 
 mod payload;
 
@@ -64,7 +65,8 @@ impl Events {
 /// [`Context::get_event_emitter`]: crate::context::Context::get_event_emitter
 /// [`Stream`]: futures::stream::Stream
 #[derive(Debug, Clone)]
-pub struct EventEmitter(Receiver<Event>);
+#[pin_project]
+pub struct EventEmitter(#[pin] Receiver<Event>);
 
 impl EventEmitter {
     /// Async recv of an event. Return `None` if the `Sender` has been dropped.
@@ -77,10 +79,10 @@ impl futures::stream::Stream for EventEmitter {
     type Item = Event;
 
     fn poll_next(
-        mut self: std::pin::Pin<&mut Self>,
+        self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<Option<Self::Item>> {
-        std::pin::Pin::new(&mut self.0).poll_next(cx)
+        self.project().0.poll_next(cx)
     }
 }