Merge #159
159: Introduce "one shots" r=D4nte a=D4nte Instead of Alice sending messages to Bob by replying on a `RequestResponse` channel, she just acks reception on the channel and open her own channel to send messages to Bob. This means we do not need to carry the channel around anymore and instead carry Bob's peer id. We take this opportunity to rename messages as per the sequence diagram. At this stage we only touch message 4 & 5 as they are the one shots messages of the protocol. Decoupling message 4 from message 3 (still current called 2) will allow the introduction of an await/async behaviour that handles message 1, 2, 3. To introduce message 1, 2, 3 we will introduced the async/await behaviour using https://github.com/comit-network/rust-libp2p-async-await First, we need to upgrade tokio and libp2p. Please note that thanks to this change, Alice now uses Bob's peer id to send message meaning that recovery is improved (as before we could not save the channel to the db). Co-authored-by: Franck Royer <franck@coblox.tech>pull/162/head
bors[bot]
3 years ago
committed by
GitHub
commit
73f89fefda
@ -0,0 +1,102 @@
|
||||
use crate::{
|
||||
monero,
|
||||
network::request_response::{OneShotCodec, Request, Response, TransferProofProtocol, TIMEOUT},
|
||||
};
|
||||
use libp2p::{
|
||||
request_response::{
|
||||
handler::RequestProtocol, ProtocolSupport, RequestResponse, RequestResponseConfig,
|
||||
RequestResponseEvent, RequestResponseMessage,
|
||||
},
|
||||
swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters},
|
||||
NetworkBehaviour, PeerId,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::{
|
||||
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)]
|
||||
pub enum OutEvent {
|
||||
Msg,
|
||||
}
|
||||
|
||||
/// A `NetworkBehaviour` that represents sending the Monero transfer proof to
|
||||
/// Bob.
|
||||
#[derive(NetworkBehaviour)]
|
||||
#[behaviour(out_event = "OutEvent", poll_method = "poll")]
|
||||
#[allow(missing_debug_implementations)]
|
||||
pub struct Behaviour {
|
||||
rr: RequestResponse<OneShotCodec<TransferProofProtocol>>,
|
||||
#[behaviour(ignore)]
|
||||
events: VecDeque<OutEvent>,
|
||||
}
|
||||
|
||||
impl Behaviour {
|
||||
pub fn send(&mut self, bob: PeerId, msg: TransferProof) {
|
||||
let msg = Request::TransferProof(Box::new(msg));
|
||||
let _id = self.rr.send_request(&bob, msg);
|
||||
}
|
||||
|
||||
fn poll(
|
||||
&mut self,
|
||||
_: &mut Context<'_>,
|
||||
_: &mut impl PollParameters,
|
||||
) -> Poll<NetworkBehaviourAction<RequestProtocol<OneShotCodec<TransferProofProtocol>>, OutEvent>>
|
||||
{
|
||||
if let Some(event) = self.events.pop_front() {
|
||||
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
|
||||
}
|
||||
|
||||
Poll::Pending
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for Behaviour {
|
||||
fn default() -> Self {
|
||||
let timeout = Duration::from_secs(TIMEOUT);
|
||||
let mut config = RequestResponseConfig::default();
|
||||
config.set_request_timeout(timeout);
|
||||
|
||||
Self {
|
||||
rr: RequestResponse::new(
|
||||
OneShotCodec::default(),
|
||||
vec![(TransferProofProtocol, ProtocolSupport::Outbound)],
|
||||
config,
|
||||
),
|
||||
events: Default::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl NetworkBehaviourEventProcess<RequestResponseEvent<Request, Response>> for Behaviour {
|
||||
fn inject_event(&mut self, event: RequestResponseEvent<Request, Response>) {
|
||||
match event {
|
||||
RequestResponseEvent::Message {
|
||||
message: RequestResponseMessage::Request { .. },
|
||||
..
|
||||
} => panic!("Alice should never get a transfer proof request from Bob"),
|
||||
RequestResponseEvent::Message {
|
||||
message: RequestResponseMessage::Response { response, .. },
|
||||
..
|
||||
} => {
|
||||
if let Response::TransferProof = response {
|
||||
self.events.push_back(OutEvent::Msg);
|
||||
}
|
||||
}
|
||||
RequestResponseEvent::InboundFailure { error, .. } => {
|
||||
error!("Inbound failure: {:?}", error);
|
||||
}
|
||||
RequestResponseEvent::OutboundFailure { error, .. } => {
|
||||
error!("Outbound failure: {:?}", error);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,97 @@
|
||||
use crate::{
|
||||
network::request_response::{OneShotCodec, Request, Response, TransferProofProtocol, TIMEOUT},
|
||||
protocol::alice::TransferProof,
|
||||
};
|
||||
use libp2p::{
|
||||
request_response::{
|
||||
handler::RequestProtocol, ProtocolSupport, RequestResponse, RequestResponseConfig,
|
||||
RequestResponseEvent, RequestResponseMessage,
|
||||
},
|
||||
swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters},
|
||||
NetworkBehaviour,
|
||||
};
|
||||
use std::{
|
||||
collections::VecDeque,
|
||||
task::{Context, Poll},
|
||||
time::Duration,
|
||||
};
|
||||
use tracing::{debug, error};
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum OutEvent {
|
||||
Msg(TransferProof),
|
||||
}
|
||||
|
||||
/// A `NetworkBehaviour` that represents receiving the transfer proof from
|
||||
/// Alice.
|
||||
#[derive(NetworkBehaviour)]
|
||||
#[behaviour(out_event = "OutEvent", poll_method = "poll")]
|
||||
#[allow(missing_debug_implementations)]
|
||||
pub struct Behaviour {
|
||||
rr: RequestResponse<OneShotCodec<TransferProofProtocol>>,
|
||||
#[behaviour(ignore)]
|
||||
events: VecDeque<OutEvent>,
|
||||
}
|
||||
|
||||
impl Behaviour {
|
||||
fn poll(
|
||||
&mut self,
|
||||
_: &mut Context<'_>,
|
||||
_: &mut impl PollParameters,
|
||||
) -> Poll<NetworkBehaviourAction<RequestProtocol<OneShotCodec<TransferProofProtocol>>, OutEvent>>
|
||||
{
|
||||
if let Some(event) = self.events.pop_front() {
|
||||
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
|
||||
}
|
||||
|
||||
Poll::Pending
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for Behaviour {
|
||||
fn default() -> Self {
|
||||
let timeout = Duration::from_secs(TIMEOUT);
|
||||
let mut config = RequestResponseConfig::default();
|
||||
config.set_request_timeout(timeout);
|
||||
|
||||
Self {
|
||||
rr: RequestResponse::new(
|
||||
OneShotCodec::default(),
|
||||
vec![(TransferProofProtocol, ProtocolSupport::Inbound)],
|
||||
config,
|
||||
),
|
||||
events: Default::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl NetworkBehaviourEventProcess<RequestResponseEvent<Request, Response>> for Behaviour {
|
||||
fn inject_event(&mut self, event: RequestResponseEvent<Request, Response>) {
|
||||
match event {
|
||||
RequestResponseEvent::Message {
|
||||
message:
|
||||
RequestResponseMessage::Request {
|
||||
request, channel, ..
|
||||
},
|
||||
..
|
||||
} => {
|
||||
if let Request::TransferProof(msg) = request {
|
||||
debug!("Received Transfer Proof");
|
||||
self.events.push_back(OutEvent::Msg(*msg));
|
||||
// Send back empty response so that the request/response protocol completes.
|
||||
let _ = self.rr.send_response(channel, Response::TransferProof);
|
||||
}
|
||||
}
|
||||
RequestResponseEvent::Message {
|
||||
message: RequestResponseMessage::Response { .. },
|
||||
..
|
||||
} => panic!("Bob should not get a Response"),
|
||||
RequestResponseEvent::InboundFailure { error, .. } => {
|
||||
error!("Inbound failure: {:?}", error);
|
||||
}
|
||||
RequestResponseEvent::OutboundFailure { error, .. } => {
|
||||
error!("Outbound failure: {:?}", error);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,48 @@
|
||||
//! A serde module that defines how we want to serialize PeerIds on the
|
||||
//! HTTP-API.
|
||||
|
||||
use libp2p::PeerId;
|
||||
use serde::{de::Error, Deserialize, Deserializer, Serializer};
|
||||
|
||||
pub fn serialize<S>(peer_id: &PeerId, serializer: S) -> Result<S::Ok, S::Error>
|
||||
where
|
||||
S: Serializer,
|
||||
{
|
||||
let string = peer_id.to_string();
|
||||
serializer.serialize_str(&string)
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub fn deserialize<'de, D>(deserializer: D) -> Result<PeerId, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
{
|
||||
let string = String::deserialize(deserializer)?;
|
||||
let peer_id = string.parse().map_err(D::Error::custom)?;
|
||||
|
||||
Ok(peer_id)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use serde::Serialize;
|
||||
use spectral::prelude::*;
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct SerializablePeerId(#[serde(with = "super")] PeerId);
|
||||
|
||||
#[test]
|
||||
fn maker_id_serializes_as_expected() {
|
||||
let peer_id = SerializablePeerId(
|
||||
"QmfUfpC2frwFvcDzpspnfZitHt5wct6n4kpG5jzgRdsxkY"
|
||||
.parse()
|
||||
.unwrap(),
|
||||
);
|
||||
|
||||
let got = serde_json::to_string(&peer_id).expect("failed to serialize peer id");
|
||||
|
||||
assert_that(&got)
|
||||
.is_equal_to(r#""QmfUfpC2frwFvcDzpspnfZitHt5wct6n4kpG5jzgRdsxkY""#.to_string());
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue