Use Rc<RefCell> instead of Arc<Mutex>

Signed-off-by: Ben Brandt <benjamin.j.brandt@gmail.com>

Author: benbrandt

rust/rpc.rs

@@ -1,5 +1,6 @@
 use std::{
     any::Any,
+    cell::RefCell,
     collections::HashMap,
     rc::Rc,
     sync::{
@@ -20,7 +21,6 @@ use futures::{
     io::BufReader,
     select_biased,
 };
-use parking_lot::Mutex;
 use serde::{Deserialize, Serialize, de::DeserializeOwned};
 use serde_json::value::RawValue;
 
@@ -29,7 +29,7 @@ use crate::{Error, StreamReceiver};
 
 pub struct RpcConnection<Local: Side, Remote: Side> {
     outgoing_tx: UnboundedSender<OutgoingMessage<Local, Remote>>,
-    pending_responses: Arc<Mutex<HashMap<i32, PendingResponse>>>,
+    pending_responses: Rc<RefCell<HashMap<i32, PendingResponse>>>,
     next_id: AtomicI32,
     broadcast: StreamBroadcast,
 }
@@ -57,7 +57,7 @@ where
         let (incoming_tx, incoming_rx) = mpsc::unbounded();
         let (outgoing_tx, outgoing_rx) = mpsc::unbounded();
 
-        let pending_responses = Arc::new(Mutex::new(HashMap::default()));
+        let pending_responses = Rc::new(RefCell::new(HashMap::default()));
         let (broadcast_tx, broadcast) = StreamBroadcast::new();
 
         let io_task = {
@@ -72,7 +72,7 @@ where
                     broadcast_tx,
                 )
                 .await;
-                pending_responses.lock().clear();
+                pending_responses.borrow_mut().clear();
                 result
             }
         };
@@ -121,7 +121,7 @@ where
     ) -> impl Future<Output = Result<Out, Error>> {
         let (tx, rx) = oneshot::channel();
         let id = self.next_id.fetch_add(1, Ordering::SeqCst);
-        self.pending_responses.lock().insert(
+        self.pending_responses.borrow_mut().insert(
             id,
             PendingResponse {
                 deserialize: |value| {
@@ -144,7 +144,7 @@ where
             })
             .is_err()
         {
-            self.pending_responses.lock().remove(&id);
+            self.pending_responses.borrow_mut().remove(&id);
         }
         async move {
             let result = rx
@@ -162,7 +162,7 @@ where
         mut outgoing_rx: UnboundedReceiver<OutgoingMessage<Local, Remote>>,
         mut outgoing_bytes: impl Unpin + AsyncWrite,
         incoming_bytes: impl Unpin + AsyncRead,
-        pending_responses: Arc<Mutex<HashMap<i32, PendingResponse>>>,
+        pending_responses: Rc<RefCell<HashMap<i32, PendingResponse>>>,
         broadcast: StreamSender,
     ) -> Result<()> {
         // TODO: Create nicer abstraction for broadcast
@@ -213,7 +213,7 @@ where
                                             broadcast.outgoing(&error_response);
                                         }
                                     }
-                                } else if let Some(pending_response) = pending_responses.lock().remove(&id) {
+                                } else if let Some(pending_response) = pending_responses.borrow_mut().remove(&id) {
                                     // Response
                                     if let Some(result_value) = message.result {
                                         broadcast.incoming_response(id, Ok(Some(result_value)));