Use event_process = false for Alice

As we do not process the event, we can just implement the needed `From`
traits.

swap/src/network/request_response.rs

@@ -128,7 +128,7 @@ where
         T: AsyncWrite + Unpin + Send,
     {
         let bytes = serde_cbor::to_vec(&res).map_err(|e| {
-            tracing::debug!("serde write_reponse error: {:?}", e);
+            tracing::debug!("serde write_response error: {:?}", e);
             io::Error::new(io::ErrorKind::InvalidData, e)
         })?;
         upgrade::write_one(io, &bytes).await?;

swap/src/protocol/alice.rs

@@ -13,7 +13,7 @@ use crate::{
     protocol::{bob::EncryptedSignature, SwapAmounts},
     seed::Seed,
 };
-use anyhow::{bail, Result};
+use anyhow::{bail, Error, Result};
 use libp2p::{
     core::Multiaddr, identity::Keypair, request_response::ResponseChannel, NetworkBehaviour, PeerId,
 };
@@ -30,6 +30,7 @@ pub use self::{
     swap_response::*,
     transfer_proof::TransferProof,
 };
+use crate::protocol::bob::SwapRequest;
 pub use execution_setup::Message3;
 
 mod encrypted_signature;
@@ -217,10 +218,18 @@ impl Builder {
 #[derive(Debug)]
 pub enum OutEvent {
     ConnectionEstablished(PeerId),
-    Request(Box<swap_response::OutEvent>),
+    SwapRequest {
+        msg: SwapRequest,
+        channel: ResponseChannel<SwapResponse>,
+    },
     ExecutionSetupDone(Result<Box<State3>>),
     TransferProofAcknowledged,
-    EncryptedSignature(Box<EncryptedSignature>),
+    EncryptedSignature {
+        msg: Box<EncryptedSignature>,
+        channel: ResponseChannel<()>,
+    },
+    ResponseSent, // Same variant is used for all messages as no processing is done
+    Failure(Error),
 }
 
 impl From<peer_tracker::OutEvent> for OutEvent {
@@ -235,7 +244,12 @@ impl From<peer_tracker::OutEvent> for OutEvent {
 
 impl From<swap_response::OutEvent> for OutEvent {
     fn from(event: swap_response::OutEvent) -> Self {
-        OutEvent::Request(Box::new(event))
+        use swap_response::OutEvent::*;
+        match event {
+            MsgReceived { msg, channel } => OutEvent::SwapRequest { msg, channel },
+            ResponseSent => OutEvent::ResponseSent,
+            Failure(err) => OutEvent::Failure(err.context("Swap Request/Response failure")),
+        }
     }
 }
 
@@ -249,16 +263,24 @@ impl From<execution_setup::OutEvent> for OutEvent {
 
 impl From<transfer_proof::OutEvent> for OutEvent {
     fn from(event: transfer_proof::OutEvent) -> Self {
+        use transfer_proof::OutEvent::*;
         match event {
-            transfer_proof::OutEvent::Acknowledged => OutEvent::TransferProofAcknowledged,
+            Acknowledged => OutEvent::TransferProofAcknowledged,
+            Failure(err) => OutEvent::Failure(err.context("Failure with Transfer Proof")),
         }
     }
 }
 
 impl From<encrypted_signature::OutEvent> for OutEvent {
     fn from(event: encrypted_signature::OutEvent) -> Self {
+        use encrypted_signature::OutEvent::*;
         match event {
-            encrypted_signature::OutEvent::Msg(msg) => OutEvent::EncryptedSignature(Box::new(msg)),
+            MsgReceived { msg, channel } => OutEvent::EncryptedSignature {
+                msg: Box::new(msg),
+                channel,
+            },
+            AckSent => OutEvent::ResponseSent,
+            Failure(err) => OutEvent::Failure(err.context("Failure with Encrypted Signature")),
         }
     }
 }

swap/src/protocol/alice/encrypted_signature.rs

@@ -2,53 +2,41 @@ use crate::{
     network::request_response::{CborCodec, EncryptedSignatureProtocol, TIMEOUT},
     protocol::bob::EncryptedSignature,
 };
+use anyhow::{anyhow, Error, Result};
 use libp2p::{
     request_response::{
-        handler::RequestProtocol, ProtocolSupport, RequestResponse, RequestResponseConfig,
+        ProtocolSupport, RequestResponse, RequestResponseConfig, RequestResponseEvent,
-        RequestResponseEvent, RequestResponseMessage,
+        RequestResponseMessage, ResponseChannel,
     },
-    swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters},
     NetworkBehaviour,
 };
-use std::{
+use std::time::Duration;
-    collections::VecDeque,
+use tracing::debug;
-    task::{Context, Poll},
-    time::Duration,
-};
-use tracing::{debug, error};
 
 #[derive(Debug)]
 pub enum OutEvent {
-    Msg(EncryptedSignature),
+    MsgReceived {
+        msg: EncryptedSignature,
+        channel: ResponseChannel<()>,
+    },
+    AckSent,
+    Failure(Error),
 }
 
 /// A `NetworkBehaviour` that represents receiving the Bitcoin encrypted
 /// signature from Bob.
 #[derive(NetworkBehaviour)]
-#[behaviour(out_event = "OutEvent", poll_method = "poll")]
+#[behaviour(out_event = "OutEvent", event_process = false)]
 #[allow(missing_debug_implementations)]
 pub struct Behaviour {
     rr: RequestResponse<CborCodec<EncryptedSignatureProtocol, EncryptedSignature, ()>>,
-    #[behaviour(ignore)]
-    events: VecDeque<OutEvent>,
 }
 
 impl Behaviour {
-    fn poll(
+    pub fn send_ack(&mut self, channel: ResponseChannel<()>) -> Result<()> {
-        &mut self,
+        self.rr
-        _: &mut Context<'_>,
+            .send_response(channel, ())
-        _: &mut impl PollParameters,
+            .map_err(|err| anyhow!("Failed to ack encrypted signature: {:?}", err))
-    ) -> Poll<
-        NetworkBehaviourAction<
-            RequestProtocol<CborCodec<EncryptedSignatureProtocol, EncryptedSignature, ()>>,
-            OutEvent,
-        >,
-    > {
-        if let Some(event) = self.events.pop_front() {
-            return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
-        }
-
-        Poll::Pending
     }
 }
 
@@ -64,41 +52,38 @@ impl Default for Behaviour {
                 vec![(EncryptedSignatureProtocol, ProtocolSupport::Inbound)],
                 config,
             ),
-            events: Default::default(),
         }
     }
 }
 
-impl NetworkBehaviourEventProcess<RequestResponseEvent<EncryptedSignature, ()>> for Behaviour {
+impl From<RequestResponseEvent<EncryptedSignature, ()>> for OutEvent {
-    fn inject_event(&mut self, event: RequestResponseEvent<EncryptedSignature, ()>) {
+    fn from(event: RequestResponseEvent<EncryptedSignature, ()>) -> Self {
         match event {
             RequestResponseEvent::Message {
+                peer,
                 message:
                     RequestResponseMessage::Request {
                         request, channel, ..
                     },
                 ..
             } => {
-                debug!("Received encrypted signature");
+                debug!("Received encrypted signature from {}", peer);
-                self.events.push_back(OutEvent::Msg(request));
+                OutEvent::MsgReceived {
-                // Send back empty response so that the request/response protocol completes.
+                    msg: request,
-                if let Err(error) = self.rr.send_response(channel, ()) {
+                    channel,
-                    error!("Failed to send Encrypted Signature ack: {:?}", error);
                 }
             }
             RequestResponseEvent::Message {
                 message: RequestResponseMessage::Response { .. },
                 ..
-            } => panic!("Alice should not get a Response"),
+            } => OutEvent::Failure(anyhow!("Alice should not get a Response")),
             RequestResponseEvent::InboundFailure { error, .. } => {
-                error!("Inbound failure: {:?}", error);
+                OutEvent::Failure(anyhow!("Inbound failure: {:?}", error))
             }
             RequestResponseEvent::OutboundFailure { error, .. } => {
-                error!("Outbound failure: {:?}", error);
+                OutEvent::Failure(anyhow!("Outbound failure: {:?}", error))
-            }
-            RequestResponseEvent::ResponseSent { .. } => {
-                debug!("Alice has sent an Message3 response to Bob");
             }
+            RequestResponseEvent::ResponseSent { .. } => OutEvent::AckSent,
         }
     }
 }

swap/src/protocol/alice/event_loop.rs

@@ -2,7 +2,7 @@ use crate::{
     network::{transport::SwapTransport, TokioExecutor},
     protocol::{
         alice::{Behaviour, OutEvent, State0, State3, SwapResponse, TransferProof},
-        bob::EncryptedSignature,
+        bob::{EncryptedSignature, SwapRequest},
     },
 };
 use anyhow::{anyhow, Context, Result};
@@ -35,7 +35,7 @@ impl<T> Default for Channels<T> {
 pub struct EventLoopHandle {
     done_execution_setup: Receiver<Result<State3>>,
     recv_encrypted_signature: Receiver<EncryptedSignature>,
-    request: Receiver<crate::protocol::alice::swap_response::OutEvent>,
+    recv_swap_request: Receiver<(SwapRequest, ResponseChannel<SwapResponse>)>,
     conn_established: Receiver<PeerId>,
     send_swap_response: Sender<(ResponseChannel<SwapResponse>, SwapResponse)>,
     start_execution_setup: Sender<(PeerId, State0)>,
@@ -70,10 +70,10 @@ impl EventLoopHandle {
             .ok_or_else(|| anyhow!("Failed to receive Bitcoin encrypted signature from Bob"))
     }
 
-    pub async fn recv_request(
+    pub async fn recv_swap_request(
         &mut self,
-    ) -> Result<crate::protocol::alice::swap_response::OutEvent> {
+    ) -> Result<(SwapRequest, ResponseChannel<SwapResponse>)> {
-        self.request
+        self.recv_swap_request
             .recv()
             .await
             .ok_or_else(|| anyhow!("Failed to receive amounts request from Bob"))
@@ -108,7 +108,7 @@ pub struct EventLoop {
     start_execution_setup: Receiver<(PeerId, State0)>,
     done_execution_setup: Sender<Result<State3>>,
     recv_encrypted_signature: Sender<EncryptedSignature>,
-    request: Sender<crate::protocol::alice::swap_response::OutEvent>,
+    recv_swap_request: Sender<(SwapRequest, ResponseChannel<SwapResponse>)>,
     conn_established: Sender<PeerId>,
     send_swap_response: Receiver<(ResponseChannel<SwapResponse>, SwapResponse)>,
     send_transfer_proof: Receiver<(PeerId, TransferProof)>,
@@ -145,7 +145,7 @@ impl EventLoop {
             start_execution_setup: start_execution_setup.receiver,
             done_execution_setup: done_execution_setup.sender,
             recv_encrypted_signature: recv_encrypted_signature.sender,
-            request: request.sender,
+            recv_swap_request: request.sender,
             conn_established: conn_established.sender,
             send_swap_response: send_swap_response.receiver,
             send_transfer_proof: send_transfer_proof.receiver,
@@ -156,7 +156,7 @@ impl EventLoop {
             start_execution_setup: start_execution_setup.sender,
             done_execution_setup: done_execution_setup.receiver,
             recv_encrypted_signature: recv_encrypted_signature.receiver,
-            request: request.receiver,
+            recv_swap_request: request.receiver,
             conn_established: conn_established.receiver,
             send_swap_response: send_swap_response.sender,
             send_transfer_proof: send_transfer_proof.sender,
@@ -174,6 +174,9 @@ impl EventLoop {
                         OutEvent::ConnectionEstablished(alice) => {
                             let _ = self.conn_established.send(alice).await;
                         }
+                        OutEvent::SwapRequest { msg, channel } => {
+                            let _ = self.recv_swap_request.send((msg, channel)).await;
+                        }
                         OutEvent::ExecutionSetupDone(res) => {
                             let _ = self.done_execution_setup.send(res.map(|state|*state)).await;
                         }
@@ -181,11 +184,16 @@ impl EventLoop {
                             trace!("Bob acknowledged transfer proof");
                             let _ = self.recv_transfer_proof_ack.send(()).await;
                         }
-                        OutEvent::EncryptedSignature(msg) => {
+                        OutEvent::EncryptedSignature{ msg, channel } => {
                             let _ = self.recv_encrypted_signature.send(*msg).await;
+                            // Send back empty response so that the request/response protocol completes.
+                            if let Err(error) = self.swarm.encrypted_signature.send_ack(channel) {
+                                error!("Failed to send Encrypted Signature ack: {:?}", error);
+                            }
                         }
-                        OutEvent::Request(event) => {
+                        OutEvent::ResponseSent => {}
-                            let _ = self.request.send(*event).await;
+                        OutEvent::Failure(err) => {
+                            error!("Communication error: {:#}", err);
                         }
                     }
                 },

swap/src/protocol/alice/steps.rs

@@ -46,13 +46,13 @@ pub async fn negotiate(
 
     let event = timeout(
         execution_params.bob_time_to_act,
-        event_loop_handle.recv_request(),
+        event_loop_handle.recv_swap_request(),
     )
     .await
     .context("Failed to receive swap request from Bob")??;
 
     event_loop_handle
-        .send_swap_response(event.channel, SwapResponse { xmr_amount })
+        .send_swap_response(event.1, SwapResponse { xmr_amount })
         .await?;
 
     let state3 = timeout(

swap/src/protocol/alice/swap_response.rs

@@ -3,27 +3,26 @@ use crate::{
     network::request_response::{CborCodec, Swap, TIMEOUT},
     protocol::bob::SwapRequest,
 };
-use anyhow::{anyhow, Result};
+use anyhow::{anyhow, Error, Result};
 use libp2p::{
     request_response::{
-        handler::RequestProtocol, ProtocolSupport, RequestResponse, RequestResponseConfig,
+        ProtocolSupport, RequestResponse, RequestResponseConfig, RequestResponseEvent,
-        RequestResponseEvent, RequestResponseMessage, ResponseChannel,
+        RequestResponseMessage, ResponseChannel,
     },
-    swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters},
     NetworkBehaviour,
 };
 use serde::{Deserialize, Serialize};
-use std::{
+use std::time::Duration;
-    collections::VecDeque,
+use tracing::debug;
-    task::{Context, Poll},
-    time::Duration,
-};
-use tracing::{debug, error};
 
 #[derive(Debug)]
-pub struct OutEvent {
+pub enum OutEvent {
-    pub msg: SwapRequest,
+    MsgReceived {
-    pub channel: ResponseChannel<SwapResponse>,
+        msg: SwapRequest,
+        channel: ResponseChannel<SwapResponse>,
+    },
+    ResponseSent,
+    Failure(Error),
 }
 
 #[derive(Copy, Clone, Debug, Serialize, Deserialize)]
@@ -31,15 +30,45 @@ pub struct SwapResponse {
     pub xmr_amount: monero::Amount,
 }
 
+impl From<RequestResponseEvent<SwapRequest, SwapResponse>> for OutEvent {
+    fn from(event: RequestResponseEvent<SwapRequest, SwapResponse>) -> Self {
+        match event {
+            RequestResponseEvent::Message {
+                peer,
+                message:
+                    RequestResponseMessage::Request {
+                        request, channel, ..
+                    },
+                ..
+            } => {
+                debug!("Received swap request from {}", peer);
+                OutEvent::MsgReceived {
+                    msg: request,
+                    channel,
+                }
+            }
+            RequestResponseEvent::Message {
+                message: RequestResponseMessage::Response { .. },
+                ..
+            } => OutEvent::Failure(anyhow!("Alice should not get a Response")),
+            RequestResponseEvent::InboundFailure { error, .. } => {
+                OutEvent::Failure(anyhow!("Inbound failure: {:?}", error))
+            }
+            RequestResponseEvent::OutboundFailure { error, .. } => {
+                OutEvent::Failure(anyhow!("Outbound failure: {:?}", error))
+            }
+            RequestResponseEvent::ResponseSent { .. } => OutEvent::ResponseSent,
+        }
+    }
+}
+
 /// A `NetworkBehaviour` that represents negotiate a swap using Swap
 /// request/response.
 #[derive(NetworkBehaviour)]
-#[behaviour(out_event = "OutEvent", poll_method = "poll")]
+#[behaviour(out_event = "OutEvent", event_process = false)]
 #[allow(missing_debug_implementations)]
 pub struct Behaviour {
     rr: RequestResponse<CborCodec<Swap, SwapRequest, SwapResponse>>,
-    #[behaviour(ignore)]
-    events: VecDeque<OutEvent>,
 }
 
 impl Behaviour {
@@ -53,23 +82,6 @@ impl Behaviour {
             .send_response(channel, msg)
             .map_err(|_| anyhow!("Sending swap response failed"))
     }
-
-    fn poll(
-        &mut self,
-        _: &mut Context<'_>,
-        _: &mut impl PollParameters,
-    ) -> Poll<
-        NetworkBehaviourAction<
-            RequestProtocol<CborCodec<Swap, SwapRequest, SwapResponse>>,
-            OutEvent,
-        >,
-    > {
-        if let Some(event) = self.events.pop_front() {
-            return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
-        }
-
-        Poll::Pending
-    }
 }
 
 impl Default for Behaviour {
@@ -85,41 +97,6 @@ impl Default for Behaviour {
                 vec![(Swap, ProtocolSupport::Inbound)],
                 config,
             ),
-            events: Default::default(),
-        }
-    }
-}
-
-impl NetworkBehaviourEventProcess<RequestResponseEvent<SwapRequest, SwapResponse>> for Behaviour {
-    fn inject_event(&mut self, event: RequestResponseEvent<SwapRequest, SwapResponse>) {
-        match event {
-            RequestResponseEvent::Message {
-                peer,
-                message:
-                    RequestResponseMessage::Request {
-                        request, channel, ..
-                    },
-                ..
-            } => {
-                debug!("Received swap request from {}", peer);
-                self.events.push_back(OutEvent {
-                    msg: request,
-                    channel,
-                })
-            }
-            RequestResponseEvent::Message {
-                message: RequestResponseMessage::Response { .. },
-                ..
-            } => panic!("Alice should not get a Response"),
-            RequestResponseEvent::InboundFailure { error, .. } => {
-                error!("Inbound failure: {:?}", error);
-            }
-            RequestResponseEvent::OutboundFailure { error, .. } => {
-                error!("Outbound failure: {:?}", error);
-            }
-            RequestResponseEvent::ResponseSent { .. } => {
-                debug!("Alice has sent a swap response to Bob");
-            }
         }
     }
 }

swap/src/protocol/alice/transfer_proof.rs

@@ -2,64 +2,41 @@ use crate::{
     monero,
     network::request_response::{CborCodec, TransferProofProtocol, TIMEOUT},
 };
+use anyhow::{anyhow, Error};
 use libp2p::{
     request_response::{
-        handler::RequestProtocol, ProtocolSupport, RequestResponse, RequestResponseConfig,
+        ProtocolSupport, RequestResponse, RequestResponseConfig, RequestResponseEvent,
-        RequestResponseEvent, RequestResponseMessage,
+        RequestResponseMessage,
     },
-    swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters},
     NetworkBehaviour, PeerId,
 };
 use serde::{Deserialize, Serialize};
-use std::{
+use std::time::Duration;
-    collections::VecDeque,
-    task::{Context, Poll},
-    time::Duration,
-};
-use tracing::error;
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct TransferProof {
     pub tx_lock_proof: monero::TransferProof,
 }
 
-#[derive(Debug, Copy, Clone)]
+#[derive(Debug)]
 pub enum OutEvent {
     Acknowledged,
+    Failure(Error),
 }
 
 /// A `NetworkBehaviour` that represents sending the Monero transfer proof to
 /// Bob.
 #[derive(NetworkBehaviour)]
-#[behaviour(out_event = "OutEvent", poll_method = "poll")]
+#[behaviour(out_event = "OutEvent", event_process = false)]
 #[allow(missing_debug_implementations)]
 pub struct Behaviour {
     rr: RequestResponse<CborCodec<TransferProofProtocol, TransferProof, ()>>,
-    #[behaviour(ignore)]
-    events: VecDeque<OutEvent>,
 }
 
 impl Behaviour {
     pub fn send(&mut self, bob: PeerId, msg: TransferProof) {
         let _id = self.rr.send_request(&bob, msg);
     }
-
-    fn poll(
-        &mut self,
-        _: &mut Context<'_>,
-        _: &mut impl PollParameters,
-    ) -> Poll<
-        NetworkBehaviourAction<
-            RequestProtocol<CborCodec<TransferProofProtocol, TransferProof, ()>>,
-            OutEvent,
-        >,
-    > {
-        if let Some(event) = self.events.pop_front() {
-            return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
-        }
-
-        Poll::Pending
-    }
 }
 
 impl Default for Behaviour {
@@ -74,31 +51,32 @@ impl Default for Behaviour {
                 vec![(TransferProofProtocol, ProtocolSupport::Outbound)],
                 config,
             ),
-            events: Default::default(),
         }
     }
 }
 
-impl NetworkBehaviourEventProcess<RequestResponseEvent<TransferProof, ()>> for Behaviour {
+impl From<RequestResponseEvent<TransferProof, ()>> for OutEvent {
-    fn inject_event(&mut self, event: RequestResponseEvent<TransferProof, ()>) {
+    fn from(event: RequestResponseEvent<TransferProof, ()>) -> Self {
         match event {
             RequestResponseEvent::Message {
                 message: RequestResponseMessage::Request { .. },
                 ..
-            } => panic!("Alice should never get a transfer proof request from Bob"),
+            } => OutEvent::Failure(anyhow!(
+                "Alice should never get a transfer proof request from Bob"
+            )),
             RequestResponseEvent::Message {
                 message: RequestResponseMessage::Response { .. },
                 ..
-            } => {
+            } => OutEvent::Acknowledged,
-                self.events.push_back(OutEvent::Acknowledged);
-            }
             RequestResponseEvent::InboundFailure { error, .. } => {
-                error!("Inbound failure: {:?}", error);
+                OutEvent::Failure(anyhow!("Inbound failure: {:?}", error))
             }
             RequestResponseEvent::OutboundFailure { error, .. } => {
-                error!("Outbound failure: {:?}", error);
+                OutEvent::Failure(anyhow!("Outbound failure: {:?}", error))
+            }
+            RequestResponseEvent::ResponseSent { .. } => {
+                OutEvent::Failure(anyhow!("Alice should not send a response"))
             }
-            RequestResponseEvent::ResponseSent { .. } => {}
         }
     }
 }