Replace `postage::oneshot` with `futures::channel::oneshot`

Antonio Scandurra created 3 years ago

This fixes an error in the randomized test that would cause the future
returned from `Worktree::share` to never finish due to a bug in `postage`
that causes its waker to not be notified upon drop.

Change summary

crates/project/src/worktree.rs | 19 ++++++++++---------
1 file changed, 10 insertions(+), 9 deletions(-)

Detailed changes

crates/project/src/worktree.rs

@@ -11,7 +11,10 @@ use client::{proto, Client, TypedEnvelope};
 use clock::ReplicaId;
 use collections::HashMap;
 use futures::{
-    channel::mpsc::{self, UnboundedSender},
+    channel::{
+        mpsc::{self, UnboundedSender},
+        oneshot,
+    },
     Stream, StreamExt,
 };
 use fuzzy::CharBag;
@@ -26,7 +29,6 @@ use language::{
 use lazy_static::lazy_static;
 use parking_lot::Mutex;
 use postage::{
-    oneshot,
     prelude::{Sink as _, Stream as _},
     watch,
 };
@@ -727,11 +729,11 @@ impl LocalWorktree {
 
     pub fn share(&mut self, project_id: u64, cx: &mut ModelContext<Worktree>) -> Task<Result<()>> {
         let register = self.register(project_id, cx);
-        let (mut share_tx, mut share_rx) = oneshot::channel();
+        let (share_tx, share_rx) = oneshot::channel();
         let (snapshots_to_send_tx, snapshots_to_send_rx) =
             smol::channel::unbounded::<LocalSnapshot>();
         if self.share.is_some() {
-            let _ = share_tx.try_send(Ok(()));
+            let _ = share_tx.send(Ok(()));
         } else {
             let rpc = self.client.clone();
             let worktree_id = cx.model_id() as u64;
@@ -756,15 +758,15 @@ impl LocalWorktree {
                                 })
                                 .await
                             {
-                                let _ = share_tx.try_send(Err(error));
+                                let _ = share_tx.send(Err(error));
                                 return Err(anyhow!("failed to send initial update worktree"));
                             } else {
-                                let _ = share_tx.try_send(Ok(()));
+                                let _ = share_tx.send(Ok(()));
                                 snapshot
                             }
                         }
                         Err(error) => {
-                            let _ = share_tx.try_send(Err(error.into()));
+                            let _ = share_tx.send(Err(error.into()));
                             return Err(anyhow!("failed to send initial update worktree"));
                         }
                     };
@@ -804,9 +806,8 @@ impl LocalWorktree {
                 });
             }
             share_rx
-                .next()
                 .await
-                .unwrap_or_else(|| Err(anyhow!("share ended")))
+                .unwrap_or_else(|_| Err(anyhow!("share ended")))
         })
     }