Introduce one shot code

To allow alice to be the requester for message 4.
This commit is contained in:
Franck Royer 2021-01-22 11:05:46 +11:00
parent 9a5e35c1bd
commit edb93624f3
No known key found for this signature in database
GPG key ID: A82ED75A8DFC50A4
3 changed files with 126 additions and 24 deletions

View file

@ -19,13 +19,14 @@ const BUF_SIZE: usize = 1024 * 1024;
// Codec for each Message and a macro that implements them. // Codec for each Message and a macro that implements them.
/// Messages Bob sends to Alice. /// Messages Bob sends to Alice.
// TODO: Remove this once network changes are done
#[allow(clippy::large_enum_variant)]
#[derive(Clone, Debug, Serialize, Deserialize)] #[derive(Clone, Debug, Serialize, Deserialize)]
pub enum BobToAlice { pub enum BobToAlice {
SwapRequest(bob::SwapRequest), SwapRequest(bob::SwapRequest),
Message0(Box<bob::Message0>), Message0(Box<bob::Message0>),
Message1(bob::Message1), Message1(bob::Message1),
Message2(bob::Message2), Message2(bob::Message2),
Message5(Message5),
} }
/// Messages Alice sends to Bob. /// Messages Alice sends to Bob.
@ -35,7 +36,19 @@ pub enum AliceToBob {
Message0(Box<alice::Message0>), Message0(Box<alice::Message0>),
Message1(Box<alice::Message1>), Message1(Box<alice::Message1>),
Message2(alice::Message2), Message2(alice::Message2),
Message5, // empty response }
/// Messages sent from one party to the other.
/// All responses are empty
#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum Request {
Message5(Message5),
}
#[derive(Clone, Debug, Serialize, Deserialize)]
/// Response are only used for acknowledgement purposes.
pub enum Response {
Message5,
} }
#[derive(Debug, Clone, Copy, Default)] #[derive(Debug, Clone, Copy, Default)]
@ -171,3 +184,92 @@ where
Ok(()) Ok(())
} }
} }
#[derive(Clone, Copy, Debug, Default)]
pub struct OneShotCodec<P> {
phantom: PhantomData<P>,
}
#[async_trait]
impl<P> RequestResponseCodec for OneShotCodec<P>
where
P: Send + Sync + Clone + ProtocolName,
{
type Protocol = P;
type Request = Request;
type Response = Response;
async fn read_request<T>(&mut self, _: &Self::Protocol, io: &mut T) -> io::Result<Self::Request>
where
T: AsyncRead + Unpin + Send,
{
debug!("enter read_request");
let message = upgrade::read_one(io, BUF_SIZE)
.await
.map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
let mut de = serde_cbor::Deserializer::from_slice(&message);
let msg = Request::deserialize(&mut de).map_err(|e| {
tracing::debug!("serde read_request error: {:?}", e);
io::Error::new(io::ErrorKind::Other, e)
})?;
Ok(msg)
}
async fn read_response<T>(
&mut self,
_: &Self::Protocol,
io: &mut T,
) -> io::Result<Self::Response>
where
T: AsyncRead + Unpin + Send,
{
debug!("enter read_response");
let message = upgrade::read_one(io, BUF_SIZE)
.await
.map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
let mut de = serde_cbor::Deserializer::from_slice(&message);
let msg = Response::deserialize(&mut de).map_err(|e| {
tracing::debug!("serde read_response error: {:?}", e);
io::Error::new(io::ErrorKind::InvalidData, e)
})?;
Ok(msg)
}
async fn write_request<T>(
&mut self,
_: &Self::Protocol,
io: &mut T,
req: Self::Request,
) -> io::Result<()>
where
T: AsyncWrite + Unpin + Send,
{
let bytes =
serde_cbor::to_vec(&req).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
upgrade::write_one(io, &bytes).await?;
Ok(())
}
async fn write_response<T>(
&mut self,
_: &Self::Protocol,
io: &mut T,
res: Self::Response,
) -> io::Result<()>
where
T: AsyncWrite + Unpin + Send,
{
debug!("enter write_response");
let bytes = serde_cbor::to_vec(&res).map_err(|e| {
tracing::debug!("serde write_reponse error: {:?}", e);
io::Error::new(io::ErrorKind::InvalidData, e)
})?;
upgrade::write_one(io, &bytes).await?;
Ok(())
}
}

View file

@ -1,5 +1,5 @@
use crate::{ use crate::{
network::request_response::{AliceToBob, BobToAlice, Codec, Message5Protocol, TIMEOUT}, network::request_response::{Message5Protocol, OneShotCodec, Request, Response, TIMEOUT},
protocol::bob::Message5, protocol::bob::Message5,
}; };
use libp2p::{ use libp2p::{
@ -27,7 +27,7 @@ pub enum OutEvent {
#[behaviour(out_event = "OutEvent", poll_method = "poll")] #[behaviour(out_event = "OutEvent", poll_method = "poll")]
#[allow(missing_debug_implementations)] #[allow(missing_debug_implementations)]
pub struct Behaviour { pub struct Behaviour {
rr: RequestResponse<Codec<Message5Protocol>>, rr: RequestResponse<OneShotCodec<Message5Protocol>>,
#[behaviour(ignore)] #[behaviour(ignore)]
events: VecDeque<OutEvent>, events: VecDeque<OutEvent>,
} }
@ -37,7 +37,8 @@ impl Behaviour {
&mut self, &mut self,
_: &mut Context<'_>, _: &mut Context<'_>,
_: &mut impl PollParameters, _: &mut impl PollParameters,
) -> Poll<NetworkBehaviourAction<RequestProtocol<Codec<Message5Protocol>>, OutEvent>> { ) -> Poll<NetworkBehaviourAction<RequestProtocol<OneShotCodec<Message5Protocol>>, OutEvent>>
{
if let Some(event) = self.events.pop_front() { if let Some(event) = self.events.pop_front() {
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event)); return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
} }
@ -54,7 +55,7 @@ impl Default for Behaviour {
Self { Self {
rr: RequestResponse::new( rr: RequestResponse::new(
Codec::default(), OneShotCodec::default(),
vec![(Message5Protocol, ProtocolSupport::Full)], vec![(Message5Protocol, ProtocolSupport::Full)],
config, config,
), ),
@ -63,8 +64,8 @@ impl Default for Behaviour {
} }
} }
impl NetworkBehaviourEventProcess<RequestResponseEvent<BobToAlice, AliceToBob>> for Behaviour { impl NetworkBehaviourEventProcess<RequestResponseEvent<Request, Response>> for Behaviour {
fn inject_event(&mut self, event: RequestResponseEvent<BobToAlice, AliceToBob>) { fn inject_event(&mut self, event: RequestResponseEvent<Request, Response>) {
match event { match event {
RequestResponseEvent::Message { RequestResponseEvent::Message {
message: message:
@ -73,12 +74,11 @@ impl NetworkBehaviourEventProcess<RequestResponseEvent<BobToAlice, AliceToBob>>
}, },
.. ..
} => { } => {
if let BobToAlice::Message5(msg) = request { let Request::Message5(msg) = request;
debug!("Received message 5"); debug!("Received message 5");
self.events.push_back(OutEvent::Msg(msg)); self.events.push_back(OutEvent::Msg(msg));
// Send back empty response so that the request/response protocol completes. // Send back empty response so that the request/response protocol completes.
self.rr.send_response(channel, AliceToBob::Message5); let _ = self.rr.send_response(channel, Response::Message5);
}
} }
RequestResponseEvent::Message { RequestResponseEvent::Message {
message: RequestResponseMessage::Response { .. }, message: RequestResponseMessage::Response { .. },

View file

@ -1,6 +1,6 @@
use crate::{ use crate::{
bitcoin::EncryptedSignature, bitcoin::EncryptedSignature,
network::request_response::{AliceToBob, BobToAlice, Codec, Message5Protocol, TIMEOUT}, network::request_response::{Message5Protocol, OneShotCodec, Request, Response, TIMEOUT},
}; };
use libp2p::{ use libp2p::{
request_response::{ request_response::{
@ -33,14 +33,14 @@ pub enum OutEvent {
#[behaviour(out_event = "OutEvent", poll_method = "poll")] #[behaviour(out_event = "OutEvent", poll_method = "poll")]
#[allow(missing_debug_implementations)] #[allow(missing_debug_implementations)]
pub struct Behaviour { pub struct Behaviour {
rr: RequestResponse<Codec<Message5Protocol>>, rr: RequestResponse<OneShotCodec<Message5Protocol>>,
#[behaviour(ignore)] #[behaviour(ignore)]
events: VecDeque<OutEvent>, events: VecDeque<OutEvent>,
} }
impl Behaviour { impl Behaviour {
pub fn send(&mut self, alice: PeerId, msg: Message5) { pub fn send(&mut self, alice: PeerId, msg: Message5) {
let msg = BobToAlice::Message5(msg); let msg = Request::Message5(msg);
let _id = self.rr.send_request(&alice, msg); let _id = self.rr.send_request(&alice, msg);
} }
@ -48,7 +48,8 @@ impl Behaviour {
&mut self, &mut self,
_: &mut Context<'_>, _: &mut Context<'_>,
_: &mut impl PollParameters, _: &mut impl PollParameters,
) -> Poll<NetworkBehaviourAction<RequestProtocol<Codec<Message5Protocol>>, OutEvent>> { ) -> Poll<NetworkBehaviourAction<RequestProtocol<OneShotCodec<Message5Protocol>>, OutEvent>>
{
if let Some(event) = self.events.pop_front() { if let Some(event) = self.events.pop_front() {
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event)); return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
} }
@ -65,7 +66,7 @@ impl Default for Behaviour {
Self { Self {
rr: RequestResponse::new( rr: RequestResponse::new(
Codec::default(), OneShotCodec::default(),
vec![(Message5Protocol, ProtocolSupport::Full)], vec![(Message5Protocol, ProtocolSupport::Full)],
config, config,
), ),
@ -74,8 +75,8 @@ impl Default for Behaviour {
} }
} }
impl NetworkBehaviourEventProcess<RequestResponseEvent<BobToAlice, AliceToBob>> for Behaviour { impl NetworkBehaviourEventProcess<RequestResponseEvent<Request, Response>> for Behaviour {
fn inject_event(&mut self, event: RequestResponseEvent<BobToAlice, AliceToBob>) { fn inject_event(&mut self, event: RequestResponseEvent<Request, Response>) {
match event { match event {
RequestResponseEvent::Message { RequestResponseEvent::Message {
message: RequestResponseMessage::Request { .. }, message: RequestResponseMessage::Request { .. },
@ -85,9 +86,8 @@ impl NetworkBehaviourEventProcess<RequestResponseEvent<BobToAlice, AliceToBob>>
message: RequestResponseMessage::Response { response, .. }, message: RequestResponseMessage::Response { response, .. },
.. ..
} => { } => {
if let AliceToBob::Message5 = response { let Response::Message5 = response;
self.events.push_back(OutEvent::Msg); self.events.push_back(OutEvent::Msg);
}
} }
RequestResponseEvent::InboundFailure { error, .. } => { RequestResponseEvent::InboundFailure { error, .. } => {
error!("Inbound failure: {:?}", error); error!("Inbound failure: {:?}", error);