Fix reusing IDs, fix closing the server event stream, apply rustfmt.

src/network/client.rs

@@ -63,7 +63,8 @@ async fn client_connection_task<StreamType, ErrorType, PacketType>(
     ErrorType: fmt::Debug,
 {
     let mut free_ids = IdAllocator::new();
-    let mut calls_in_progress = BTreeMap::<u32, oneshot::Sender<Result<PacketType, RPCError<ErrorType>>>>::new();
+    let mut calls_in_progress =
+        BTreeMap::<u32, oneshot::Sender<Result<PacketType, RPCError<ErrorType>>>>::new();
 
     // TODO: Refactor this function.
 
@@ -82,10 +83,14 @@ async fn client_connection_task<StreamType, ErrorType, PacketType>(
                                     // We received a response to a call - find the call and forward
                                     // the reply.
                                     if let Some(call) = calls_in_progress.remove(&deserialized.id) {
+                                        free_ids.free(deserialized.id);
                                         call.send(Ok(deserialized.payload)).ok();
                                     } else {
                                         // TODO: Use proper logging functionality.
-                                        eprintln!("Received a reply for an unknown call ({:?})!", deserialized.id);
+                                        eprintln!(
+                                            "Received a reply for an unknown call ({:?})!",
+                                            deserialized.id
+                                        );
                                     }
                                 }
                             }
@@ -159,6 +164,8 @@ async fn client_connection_task<StreamType, ErrorType, PacketType>(
         };
     }
 
+    drop(event_send);
+
     // Return errors for all pending packets as there will be no responses.
     for (_, result) in calls_in_progress {
         result.send(Err(RPCError::Closed)).ok();
@@ -224,9 +231,9 @@ impl<PacketType, ErrorType> RPCInterface<PacketType, ErrorType> {
 impl<PacketType, ErrorType> Drop for RPCInterface<PacketType, ErrorType> {
     fn drop(&mut self) {
         // Stop the connection loop and close the connection.
-        if let Err(_) = self.cmd_send.send(ClientTaskCommand::Close) {
-            panic!("could not send close signal to connection task");
-        }
+        // send() only fails if the other end has already been closed - we must
+        // not unwrap() or expect() because then nested panics might occur.
+        self.cmd_send.send(ClientTaskCommand::Close).ok();
     }
 }