refactor response processing

helix-dap/src/transport.rs

@@ -1,6 +1,6 @@
 use crate::{Error, Result};
 use anyhow::Context;
-use log::{error, info};
+use log::{error, info, warn};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use std::collections::HashMap;
@@ -162,54 +162,50 @@ impl Transport {
         Ok(())
     }
 
-    async fn process_server_message(
+    fn process_response(res: Response) -> Result<Response> {
-        &self,
+        match res.success {
-        client_tx: &UnboundedSender<Payload>,
-        msg: Payload,
-    ) -> Result<()> {
-        match msg {
-            Payload::Response(Response {
-                ref success,
-                ref seq,
-                request_seq,
-                ref command,
-                ref message,
-                ref body,
-                ..
-            }) => {
-                let result = match success {
             true => {
-                        info!("<- DAP success ({}, in response to {})", seq, request_seq);
+                info!(
-                        if let Payload::Response(val) = msg {
+                    "<- DAP success ({}, in response to {})",
-                            Ok(val)
+                    res.seq, res.request_seq
-                        } else {
+                );
-                            unreachable!();
+
-                        }
+                Ok(res)
             }
             false => {
                 error!(
                     "<- DAP error {:?} ({:?}) for command #{} {}",
-                            message, body, request_seq, command
+                    res.message, res.body, res.request_seq, res.command
                 );
 
-                        Err(Error::Other(anyhow::format_err!("{:?}", body)))
+                Err(Error::Other(anyhow::format_err!("{:?}", res.body)))
+            }
+        }
     }
-                };
 
-                let tx = self
+    async fn process_server_message(
-                    .pending_requests
+        &self,
-                    .lock()
+        client_tx: &UnboundedSender<Payload>,
-                    .await
+        msg: Payload,
-                    .remove(&request_seq)
+    ) -> Result<()> {
-                    .expect("pending_request with id not found!");
+        match msg {
+            Payload::Response(res) => {
+                let request_seq = res.request_seq;
+                let tx = self.pending_requests.lock().await.remove(&request_seq);
 
-                match tx.send(result).await {
+                match tx {
+                    Some(tx) => match tx.send(Self::process_response(res)).await {
                         Ok(_) => (),
                         Err(_) => error!(
                             "Tried sending response into a closed channel (id={:?}), original request likely timed out",
                             request_seq
                         ),
-                };
+                    }
+                    None => {
+                        warn!("Response to nonexistent request #{}", res.request_seq);
+                        client_tx.send(Payload::Response(res)).expect("Failed to send");
+                    }
+                }
 
                 Ok(())
             }