Replace flume with async_channel in docs

This is mostly a 1:1 replacement, except for the fact that the same_channel
api is missing from async_channel. So I replaced it with some ugly code
that uses the fact that a async_channel Sender or Receiver is just an Arc<Channel>.

To be removed if/when smol-rs/async-channel#98 is
merged, but until then I think it is fine.

Author: rklaehn

Cargo.lock

@@ -518,6 +518,93 @@ impl<T: Send + Sync + 'static> ProgressSender for FlumeProgressSender<T> {
     }
 }
 
+/// A progress sender that uses a flume channel.
+pub struct AsyncChannelProgressSender<T> {
+    sender: async_channel::Sender<T>,
+    id: std::sync::Arc<std::sync::atomic::AtomicU64>,
+}
+
+impl<T> std::fmt::Debug for AsyncChannelProgressSender<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AsyncChannelProgressSender")
+            .field("id", &self.id)
+            .field("sender", &self.sender)
+            .finish()
+    }
+}
+
+impl<T> Clone for AsyncChannelProgressSender<T> {
+    fn clone(&self) -> Self {
+        Self {
+            sender: self.sender.clone(),
+            id: self.id.clone(),
+        }
+    }
+}
+
+impl<T> AsyncChannelProgressSender<T> {
+    /// Create a new progress sender from a flume sender.
+    pub fn new(sender: async_channel::Sender<T>) -> Self {
+        Self {
+            sender,
+            id: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0)),
+        }
+    }
+
+    /// Returns true if `other` sends on the same `flume` channel as `self`.
+    pub fn same_channel(&self, other: &AsyncChannelProgressSender<T>) -> bool {
+        same_channel(&self.sender, &other.sender)
+    }
+}
+
+fn get_as_ptr<T>(value: &T) -> Option<usize> {
+    use std::mem;
+    if mem::size_of::<T>() == std::mem::size_of::<usize>()
+        && mem::align_of::<T>() == mem::align_of::<usize>()
+    {
+        // Safe only if size and alignment requirements are met
+        unsafe { Some(mem::transmute_copy(value)) }
+    } else {
+        None
+    }
+}
+
+fn same_channel<T>(a: &async_channel::Sender<T>, b: &async_channel::Sender<T>) -> bool {
+    get_as_ptr(a).unwrap() == get_as_ptr(b).unwrap()
+}
+
+impl<T> IdGenerator for AsyncChannelProgressSender<T> {
+    fn new_id(&self) -> u64 {
+        self.id.fetch_add(1, std::sync::atomic::Ordering::SeqCst)
+    }
+}
+
+impl<T: Send + Sync + 'static> ProgressSender for AsyncChannelProgressSender<T> {
+    type Msg = T;
+
+    async fn send(&self, msg: Self::Msg) -> std::result::Result<(), ProgressSendError> {
+        self.sender
+            .send(msg)
+            .await
+            .map_err(|_| ProgressSendError::ReceiverDropped)
+    }
+
+    fn try_send(&self, msg: Self::Msg) -> std::result::Result<(), ProgressSendError> {
+        match self.sender.try_send(msg) {
+            Ok(_) => Ok(()),
+            Err(async_channel::TrySendError::Full(_)) => Ok(()),
+            Err(async_channel::TrySendError::Closed(_)) => Err(ProgressSendError::ReceiverDropped),
+        }
+    }
+
+    fn blocking_send(&self, msg: Self::Msg) -> std::result::Result<(), ProgressSendError> {
+        match self.sender.send_blocking(msg) {
+            Ok(_) => Ok(()),
+            Err(_) => Err(ProgressSendError::ReceiverDropped),
+        }
+    }
+}
+
 /// An error that can occur when sending progress messages.
 ///
 /// Really the only error that can occur is if the receiver is dropped.

iroh-blobs/src/util/progress.rs

@@ -16,11 +16,11 @@ workspace = true
 
 [dependencies]
 anyhow = "1"
+async-channel = "2.3.1"
 blake3 = { package = "iroh-blake3", version = "1.4.5"}
 bytes = { version = "1.4", features = ["serde"] }
 derive_more = { version = "1.0.0-beta.6", features = ["debug", "deref", "display", "from", "try_into", "into", "as_ref"] }
 ed25519-dalek = { version = "2.0.0", features = ["serde", "rand_core"] }
-flume = "0.11"
 futures-buffered = "0.2.4"
 futures-lite = "2.3.0"
 futures-util = { version = "0.3.25" }

iroh-docs/Cargo.toml

@@ -61,12 +61,12 @@ enum Action {
     #[display("ListAuthors")]
     ListAuthors {
         #[debug("reply")]
-        reply: flume::Sender<Result<AuthorId>>,
+        reply: async_channel::Sender<Result<AuthorId>>,
     },
     #[display("ListReplicas")]
     ListReplicas {
         #[debug("reply")]
-        reply: flume::Sender<Result<(NamespaceId, CapabilityKind)>>,
+        reply: async_channel::Sender<Result<(NamespaceId, CapabilityKind)>>,
     },
     #[display("ContentHashes")]
     ContentHashes {
@@ -108,12 +108,12 @@ enum ReplicaAction {
         reply: oneshot::Sender<Result<()>>,
     },
     Subscribe {
-        sender: flume::Sender<Event>,
+        sender: async_channel::Sender<Event>,
         #[debug("reply")]
         reply: oneshot::Sender<Result<()>>,
     },
     Unsubscribe {
-        sender: flume::Sender<Event>,
+        sender: async_channel::Sender<Event>,
         #[debug("reply")]
         reply: oneshot::Sender<Result<()>>,
     },
@@ -166,7 +166,7 @@ enum ReplicaAction {
     },
     GetMany {
         query: Query,
-        reply: flume::Sender<Result<SignedEntry>>,
+        reply: async_channel::Sender<Result<SignedEntry>>,
     },
     DropReplica {
         reply: oneshot::Sender<Result<()>>,
@@ -222,7 +222,7 @@ struct OpenReplica {
 /// [`SyncHandle::drop`] will not block.
 #[derive(Debug, Clone)]
 pub struct SyncHandle {
-    tx: flume::Sender<Action>,
+    tx: async_channel::Sender<Action>,
     join_handle: Arc<Option<JoinHandle<()>>>,
 }
 
@@ -232,7 +232,7 @@ pub struct OpenOpts {
     /// Set to true to set sync state to true.
     pub sync: bool,
     /// Optionally subscribe to replica events.
-    pub subscribe: Option<flume::Sender<Event>>,
+    pub subscribe: Option<async_channel::Sender<Event>>,
 }
 impl OpenOpts {
     /// Set sync state to true.
@@ -241,7 +241,7 @@ impl OpenOpts {
         self
     }
     /// Subscribe to replica events.
-    pub fn subscribe(mut self, subscribe: flume::Sender<Event>) -> Self {
+    pub fn subscribe(mut self, subscribe: async_channel::Sender<Event>) -> Self {
         self.subscribe = Some(subscribe);
         self
     }
@@ -255,7 +255,7 @@ impl SyncHandle {
         content_status_callback: Option<ContentStatusCallback>,
         me: String,
     ) -> SyncHandle {
-        let (action_tx, action_rx) = flume::bounded(ACTION_CAP);
+        let (action_tx, action_rx) = async_channel::bounded(ACTION_CAP);
         let actor = Actor {
             store,
             states: Default::default(),
@@ -298,7 +298,7 @@ impl SyncHandle {
     pub async fn subscribe(
         &self,
         namespace: NamespaceId,
-        sender: flume::Sender<Event>,
+        sender: async_channel::Sender<Event>,
     ) -> Result<()> {
         let (reply, rx) = oneshot::channel();
         self.send_replica(namespace, ReplicaAction::Subscribe { sender, reply })
@@ -309,7 +309,7 @@ impl SyncHandle {
     pub async fn unsubscribe(
         &self,
         namespace: NamespaceId,
-        sender: flume::Sender<Event>,
+        sender: async_channel::Sender<Event>,
     ) -> Result<()> {
         let (reply, rx) = oneshot::channel();
         self.send_replica(namespace, ReplicaAction::Unsubscribe { sender, reply })
@@ -435,7 +435,7 @@ impl SyncHandle {
         &self,
         namespace: NamespaceId,
         query: Query,
-        reply: flume::Sender<Result<SignedEntry>>,
+        reply: async_channel::Sender<Result<SignedEntry>>,
     ) -> Result<()> {
         let action = ReplicaAction::GetMany { query, reply };
         self.send_replica(namespace, action).await?;
@@ -489,13 +489,13 @@ impl SyncHandle {
         Ok(store)
     }
 
-    pub async fn list_authors(&self, reply: flume::Sender<Result<AuthorId>>) -> Result<()> {
+    pub async fn list_authors(&self, reply: async_channel::Sender<Result<AuthorId>>) -> Result<()> {
         self.send(Action::ListAuthors { reply }).await
     }
 
     pub async fn list_replicas(
         &self,
-        reply: flume::Sender<Result<(NamespaceId, CapabilityKind)>>,
+        reply: async_channel::Sender<Result<(NamespaceId, CapabilityKind)>>,
     ) -> Result<()> {
         self.send(Action::ListReplicas { reply }).await
     }
@@ -566,7 +566,7 @@ impl SyncHandle {
 
     async fn send(&self, action: Action) -> Result<()> {
         self.tx
-            .send_async(action)
+            .send(action)
             .await
             .context("sending to iroh_docs actor failed")?;
         Ok(())
@@ -581,7 +581,10 @@ impl Drop for SyncHandle {
     fn drop(&mut self) {
         // this means we're dropping the last reference
         if let Some(handle) = Arc::get_mut(&mut self.join_handle) {
-            self.tx.send(Action::Shutdown { reply: None }).ok();
+            // this call is the reason tx can not be a tokio mpsc channel.
+            // we have no control about where drop is called, yet tokio send_blocking panics
+            // when called from inside a tokio runtime.
+            self.tx.send_blocking(Action::Shutdown { reply: None }).ok();
             let handle = handle.take().expect("this can only run once");
             if let Err(err) = handle.join() {
                 warn!(?err, "Failed to join sync actor");
@@ -593,7 +596,7 @@ impl Drop for SyncHandle {
 struct Actor {
     store: Store,
     states: OpenReplicas,
-    action_rx: flume::Receiver<Action>,
+    action_rx: async_channel::Receiver<Action>,
     content_status_callback: Option<ContentStatusCallback>,
     tasks: JoinSet<()>,
 }
@@ -619,10 +622,10 @@ impl Actor {
                     }
                     continue;
                 }
-                action = self.action_rx.recv_async() => {
+                action = self.action_rx.recv() => {
                     match action {
                         Ok(action) => action,
-                        Err(flume::RecvError::Disconnected) => {
+                        Err(async_channel::RecvError) => {
                             debug!("action channel disconnected");
                             break None;
                         }
@@ -979,17 +982,14 @@ impl OpenReplicas {
 }
 
 async fn iter_to_channel_async<T: Send + 'static>(
-    channel: flume::Sender<Result<T>>,
+    channel: async_channel::Sender<Result<T>>,
     iter: Result<impl Iterator<Item = Result<T>>>,
 ) -> Result<(), SendReplyError> {
     match iter {
-        Err(err) => channel
-            .send_async(Err(err))
-            .await
-            .map_err(send_reply_error)?,
+        Err(err) => channel.send(Err(err)).await.map_err(send_reply_error)?,
         Ok(iter) => {
             for item in iter {
-                channel.send_async(item).await.map_err(send_reply_error)?;
+                channel.send(item).await.map_err(send_reply_error)?;
             }
         }
     }
@@ -1032,10 +1032,10 @@ mod tests {
         let id = namespace.id();
         sync.import_namespace(namespace.into()).await?;
         sync.open(id, Default::default()).await?;
-        let (tx, rx) = flume::bounded(10);
+        let (tx, rx) = async_channel::bounded(10);
         sync.subscribe(id, tx).await?;
         sync.close(id).await?;
-        assert!(rx.recv_async().await.is_err());
+        assert!(rx.recv().await.is_err());
         Ok(())
     }
 }

iroh-docs/src/actor.rs

@@ -170,15 +170,15 @@ impl Engine {
 
         // Subscribe to insert events from the replica.
         let a = {
-            let (s, r) = flume::bounded(SUBSCRIBE_CHANNEL_CAP);
+            let (s, r) = async_channel::bounded(SUBSCRIBE_CHANNEL_CAP);
             this.sync.subscribe(namespace, s).await?;
-            r.into_stream()
-                .map(move |ev| LiveEvent::from_replica_event(ev, &content_status_cb))
+            Box::pin(r).map(move |ev| LiveEvent::from_replica_event(ev, &content_status_cb))
         };
 
         // Subscribe to events from the [`live::Actor`].
         let b = {
-            let (s, r) = flume::bounded(SUBSCRIBE_CHANNEL_CAP);
+            let (s, r) = async_channel::bounded(SUBSCRIBE_CHANNEL_CAP);
+            let r = Box::pin(r);
             let (reply, reply_rx) = oneshot::channel();
             this.to_live_actor
                 .send(ToLiveActor::Subscribe {
@@ -188,7 +188,7 @@ impl Engine {
                 })
                 .await?;
             reply_rx.await??;
-            r.into_stream().map(|event| Ok(LiveEvent::from(event)))
+            r.map(|event| Ok(LiveEvent::from(event)))
         };
 
         Ok(a.or(b))