generalize to support requests. untested

1 files changed, 22 insertions, 16 deletions

diff --git a/src/transaction/mod.rs b/src/transaction/mod.rs
index 2953cb7..3be9544 100644
--- a/src/transaction/mod.rs
+++ b/src/transaction/mod.rs
@@ -1,5 +1,5 @@
 use crate::{
-    encoding::{headers::CSeq, request::Request, response::Response},
+    encoding::{headers::CSeq, request::Request, response::Response, Message},
     transport::Transport,
 };
 use anyhow::{anyhow, Result};
@@ -27,20 +27,26 @@ impl<T: Transport> TransactionUser<T> {
         }
     }
 
-    pub async fn process_responses(&self) -> Result<()> {
-        let resp = self.transport.recv().await?;
-        let cseq = resp
-            .headers
-            .get()
-            .ok_or(anyhow!("response cseq missing"))??;
-        self.pending_requests
-            .write()
-            .await
-            .get_mut(&cseq)
-            .ok_or(anyhow!("message was not requested"))?
-            .send(resp)
-            .await?;
-        Ok(())
+    pub async fn process_incoming(&self) -> Result<Request> {
+        loop {
+            let mesg = self.transport.recv().await?;
+            match mesg {
+                Message::Request(req) => break Ok(req),
+                Message::Response(resp) => {
+                    let cseq = resp
+                        .headers
+                        .get()
+                        .ok_or(anyhow!("response cseq missing"))??;
+                    self.pending_requests
+                        .write()
+                        .await
+                        .get_mut(&cseq)
+                        .ok_or(anyhow!("message was not requested"))?
+                        .send(resp)
+                        .await?;
+                }
+            }
+        }
     }
 
     pub async fn transact(&self, mut request: Request) -> Result<mpsc::Receiver<Response>> {
@@ -50,7 +56,7 @@ impl<T: Transport> TransactionUser<T> {
 
         let (tx, rx) = channel(4);
 
-        self.transport.send(request).await?;
+        self.transport.send(Message::Request(request)).await?;
         self.pending_requests.write().await.insert(cseq, tx);
 
         Ok(rx)