use futures::{channel::{mpsc, oneshot}, prelude::*};
use libp2p::{
core::{
connection::{ConnectionId, ListenerId},
ConnectedPoint, Multiaddr, PeerId,
},
request_response::{
RequestResponse, RequestResponseCodec, RequestResponseConfig, RequestResponseEvent,
RequestResponseMessage, ResponseChannel, ProtocolSupport
},
swarm::{
protocols_handler::multi::MultiHandler, NetworkBehaviour, NetworkBehaviourAction,
PollParameters, ProtocolsHandler,
},
};
use std::{
borrow::Cow, collections::{hash_map::Entry, HashMap}, convert::TryFrom as _, io, iter,
pin::Pin, task::{Context, Poll}, time::Duration,
};
use wasm_timer::Instant;
use crate::ReputationChange;
pub use libp2p::request_response::{InboundFailure, OutboundFailure, RequestId};
#[derive(Debug, Clone)]
pub struct ProtocolConfig {
pub name: Cow<'static, str>,
pub max_request_size: u64,
pub max_response_size: u64,
pub request_timeout: Duration,
pub inbound_queue: Option<mpsc::Sender<IncomingRequest>>,
}
#[derive(Debug)]
pub struct IncomingRequest {
pub peer: PeerId,
pub payload: Vec<u8>,
pub pending_response: oneshot::Sender<OutgoingResponse>,
}
#[derive(Debug)]
pub struct OutgoingResponse {
pub result: Result<Vec<u8>, ()>,
pub reputation_changes: Vec<ReputationChange>,
}
#[derive(Debug)]
pub enum Event {
InboundRequest {
peer: PeerId,
protocol: Cow<'static, str>,
result: Result<Duration, ResponseFailure>,
},
RequestFinished {
peer: PeerId,
protocol: Cow<'static, str>,
duration: Duration,
result: Result<(), RequestFailure>
},
ReputationChanges {
peer: PeerId,
changes: Vec<ReputationChange>,
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct ProtocolRequestId {
protocol: Cow<'static, str>,
request_id: RequestId,
}
impl From<(Cow<'static, str>, RequestId)> for ProtocolRequestId {
fn from((protocol, request_id): (Cow<'static, str>, RequestId)) -> Self {
Self { protocol, request_id }
}
}
pub enum IfDisconnected {
TryConnect,
ImmediateError,
}
impl IfDisconnected {
pub fn should_connect(self) -> bool {
match self {
Self::TryConnect => true,
Self::ImmediateError => false,
}
}
}
pub struct RequestResponsesBehaviour {
protocols: HashMap<
Cow<'static, str>,
(RequestResponse<GenericCodec>, Option<mpsc::Sender<IncomingRequest>>)
>,
pending_requests: HashMap<
ProtocolRequestId,
(Instant, oneshot::Sender<Result<Vec<u8>, RequestFailure>>),
>,
pending_responses: stream::FuturesUnordered<
Pin<Box<dyn Future<Output = Option<RequestProcessingOutcome>> + Send>>
>,
pending_responses_arrival_time: HashMap<ProtocolRequestId, Instant>,
}
struct RequestProcessingOutcome {
peer: PeerId,
request_id: RequestId,
protocol: Cow<'static, str>,
inner_channel: ResponseChannel<Result<Vec<u8>, ()>>,
response: OutgoingResponse,
}
impl RequestResponsesBehaviour {
pub fn new(list: impl Iterator<Item = ProtocolConfig>) -> Result<Self, RegisterError> {
let mut protocols = HashMap::new();
for protocol in list {
let mut cfg = RequestResponseConfig::default();
cfg.set_connection_keep_alive(Duration::from_secs(10));
cfg.set_request_timeout(protocol.request_timeout);
let protocol_support = if protocol.inbound_queue.is_some() {
ProtocolSupport::Full
} else {
ProtocolSupport::Outbound
};
let rq_rp = RequestResponse::new(GenericCodec {
max_request_size: protocol.max_request_size,
max_response_size: protocol.max_response_size,
}, iter::once((protocol.name.as_bytes().to_vec(), protocol_support)), cfg);
match protocols.entry(protocol.name) {
Entry::Vacant(e) => e.insert((rq_rp, protocol.inbound_queue)),
Entry::Occupied(e) =>
return Err(RegisterError::DuplicateProtocol(e.key().clone())),
};
}
Ok(Self {
protocols,
pending_requests: Default::default(),
pending_responses: Default::default(),
pending_responses_arrival_time: Default::default(),
})
}
pub fn send_request(
&mut self,
target: &PeerId,
protocol_name: &str,
request: Vec<u8>,
pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
connect: IfDisconnected,
) {
if let Some((protocol, _)) = self.protocols.get_mut(protocol_name) {
if protocol.is_connected(target) || connect.should_connect() {
let request_id = protocol.send_request(target, request);
let prev_req_id = self.pending_requests.insert(
(protocol_name.to_string().into(), request_id).into(),
(Instant::now(), pending_response),
);
debug_assert!(prev_req_id.is_none(), "Expect request id to be unique.");
} else {
if pending_response.send(Err(RequestFailure::NotConnected)).is_err() {
log::debug!(
target: "sub-libp2p",
"Not connected to peer {:?}. At the same time local \
node is no longer interested in the result.",
target,
);
};
}
} else {
if pending_response.send(Err(RequestFailure::UnknownProtocol)).is_err() {
log::debug!(
target: "sub-libp2p",
"Unknown protocol {:?}. At the same time local \
node is no longer interested in the result.",
protocol_name,
);
};
}
}
}
impl NetworkBehaviour for RequestResponsesBehaviour {
type ProtocolsHandler = MultiHandler<
String,
<RequestResponse<GenericCodec> as NetworkBehaviour>::ProtocolsHandler,
>;
type OutEvent = Event;
fn new_handler(&mut self) -> Self::ProtocolsHandler {
let iter = self.protocols.iter_mut()
.map(|(p, (r, _))| (p.to_string(), NetworkBehaviour::new_handler(r)));
MultiHandler::try_from_iter(iter)
.expect("Protocols are in a HashMap and there can be at most one handler per \
protocol name, which is the only possible error; qed")
}
fn addresses_of_peer(&mut self, _: &PeerId) -> Vec<Multiaddr> {
Vec::new()
}
fn inject_connection_established(
&mut self,
peer_id: &PeerId,
conn: &ConnectionId,
endpoint: &ConnectedPoint,
) {
for (p, _) in self.protocols.values_mut() {
NetworkBehaviour::inject_connection_established(p, peer_id, conn, endpoint)
}
}
fn inject_connected(&mut self, peer_id: &PeerId) {
for (p, _) in self.protocols.values_mut() {
NetworkBehaviour::inject_connected(p, peer_id)
}
}
fn inject_connection_closed(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
for (p, _) in self.protocols.values_mut() {
NetworkBehaviour::inject_connection_closed(p, peer_id, conn, endpoint)
}
}
fn inject_disconnected(&mut self, peer_id: &PeerId) {
for (p, _) in self.protocols.values_mut() {
NetworkBehaviour::inject_disconnected(p, peer_id)
}
}
fn inject_addr_reach_failure(
&mut self,
peer_id: Option<&PeerId>,
addr: &Multiaddr,
error: &dyn std::error::Error
) {
for (p, _) in self.protocols.values_mut() {
NetworkBehaviour::inject_addr_reach_failure(p, peer_id, addr, error)
}
}
fn inject_event(
&mut self,
peer_id: PeerId,
connection: ConnectionId,
(p_name, event): <Self::ProtocolsHandler as ProtocolsHandler>::OutEvent,
) {
if let Some((proto, _)) = self.protocols.get_mut(&*p_name) {
return proto.inject_event(peer_id, connection, event)
}
log::warn!(target: "sub-libp2p",
"inject_node_event: no request-response instance registered for protocol {:?}",
p_name)
}
fn inject_new_external_addr(&mut self, addr: &Multiaddr) {
for (p, _) in self.protocols.values_mut() {
NetworkBehaviour::inject_new_external_addr(p, addr)
}
}
fn inject_expired_listen_addr(&mut self, addr: &Multiaddr) {
for (p, _) in self.protocols.values_mut() {
NetworkBehaviour::inject_expired_listen_addr(p, addr)
}
}
fn inject_dial_failure(&mut self, peer_id: &PeerId) {
for (p, _) in self.protocols.values_mut() {
NetworkBehaviour::inject_dial_failure(p, peer_id)
}
}
fn inject_new_listen_addr(&mut self, addr: &Multiaddr) {
for (p, _) in self.protocols.values_mut() {
NetworkBehaviour::inject_new_listen_addr(p, addr)
}
}
fn inject_listener_error(&mut self, id: ListenerId, err: &(dyn std::error::Error + 'static)) {
for (p, _) in self.protocols.values_mut() {
NetworkBehaviour::inject_listener_error(p, id, err)
}
}
fn inject_listener_closed(&mut self, id: ListenerId, reason: Result<(), &io::Error>) {
for (p, _) in self.protocols.values_mut() {
NetworkBehaviour::inject_listener_closed(p, id, reason)
}
}
fn poll(
&mut self,
cx: &mut Context,
params: &mut impl PollParameters,
) -> Poll<
NetworkBehaviourAction<
<Self::ProtocolsHandler as ProtocolsHandler>::InEvent,
Self::OutEvent,
>,
> {
'poll_all: loop {
while let Poll::Ready(Some(outcome)) = self.pending_responses.poll_next_unpin(cx) {
let RequestProcessingOutcome {
peer,
request_id,
protocol: protocol_name,
inner_channel,
response: OutgoingResponse {
result,
reputation_changes,
},
} = match outcome {
Some(outcome) => outcome,
None => continue,
};
if let Ok(payload) = result {
if let Some((protocol, _)) = self.protocols.get_mut(&*protocol_name) {
if let Err(_) = protocol.send_response(inner_channel, Ok(payload)) {
log::debug!(
target: "sub-libp2p",
"Failed to send response for {:?} on protocol {:?} due to a \
timeout or due to the connection to the peer being closed. \
Dropping response",
request_id, protocol_name,
);
}
}
}
if !reputation_changes.is_empty() {
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(
Event::ReputationChanges{
peer,
changes: reputation_changes,
},
));
}
}
for (protocol, (behaviour, resp_builder)) in &mut self.protocols {
while let Poll::Ready(ev) = behaviour.poll(cx, params) {
let ev = match ev {
NetworkBehaviourAction::GenerateEvent(ev) => ev,
NetworkBehaviourAction::DialAddress { address } => {
log::error!("The request-response isn't supposed to start dialing peers");
return Poll::Ready(NetworkBehaviourAction::DialAddress { address })
}
NetworkBehaviourAction::DialPeer { peer_id, condition } => {
return Poll::Ready(NetworkBehaviourAction::DialPeer {
peer_id,
condition,
})
}
NetworkBehaviourAction::NotifyHandler {
peer_id,
handler,
event,
} => {
return Poll::Ready(NetworkBehaviourAction::NotifyHandler {
peer_id,
handler,
event: ((*protocol).to_string(), event),
})
}
NetworkBehaviourAction::ReportObservedAddr { address, score } => {
return Poll::Ready(NetworkBehaviourAction::ReportObservedAddr {
address, score,
})
}
};
match ev {
RequestResponseEvent::Message {
peer,
message: RequestResponseMessage::Request { request_id, request, channel, .. },
} => {
self.pending_responses_arrival_time.insert(
(protocol.clone(), request_id.clone()).into(),
Instant::now(),
);
let (tx, rx) = oneshot::channel();
if let Some(resp_builder) = resp_builder {
let _ = resp_builder.try_send(IncomingRequest {
peer: peer.clone(),
payload: request,
pending_response: tx,
});
} else {
debug_assert!(false, "Received message on outbound-only protocol.");
}
let protocol = protocol.clone();
self.pending_responses.push(Box::pin(async move {
if let Ok(response) = rx.await {
Some(RequestProcessingOutcome {
peer, request_id, protocol, inner_channel: channel, response
})
} else {
None
}
}));
continue 'poll_all;
}
RequestResponseEvent::Message {
peer,
message: RequestResponseMessage::Response {
request_id,
response,
},
..
} => {
let (started, delivered) = match self.pending_requests.remove(
&(protocol.clone(), request_id).into(),
) {
Some((started, pending_response)) => {
let delivered = pending_response.send(
response.map_err(|()| RequestFailure::Refused),
).map_err(|_| RequestFailure::Obsolete);
(started, delivered)
}
None => {
log::warn!(
target: "sub-libp2p",
"Received `RequestResponseEvent::Message` with unexpected request id {:?}",
request_id,
);
debug_assert!(false);
continue;
}
};
let out = Event::RequestFinished {
peer,
protocol: protocol.clone(),
duration: started.elapsed(),
result: delivered,
};
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(out));
}
RequestResponseEvent::OutboundFailure {
peer,
request_id,
error,
..
} => {
let started = match self.pending_requests.remove(&(protocol.clone(), request_id).into()) {
Some((started, pending_response)) => {
if pending_response.send(
Err(RequestFailure::Network(error.clone())),
).is_err() {
log::debug!(
target: "sub-libp2p",
"Request with id {:?} failed. At the same time local \
node is no longer interested in the result.",
request_id,
);
}
started
}
None => {
log::warn!(
target: "sub-libp2p",
"Received `RequestResponseEvent::Message` with unexpected request id {:?}",
request_id,
);
debug_assert!(false);
continue;
}
};
let out = Event::RequestFinished {
peer,
protocol: protocol.clone(),
duration: started.elapsed(),
result: Err(RequestFailure::Network(error)),
};
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(out));
}
RequestResponseEvent::InboundFailure { request_id, peer, error, .. } => {
self.pending_responses_arrival_time.remove(
&(protocol.clone(), request_id).into(),
);
let out = Event::InboundRequest {
peer,
protocol: protocol.clone(),
result: Err(ResponseFailure::Network(error)),
};
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(out));
}
RequestResponseEvent::ResponseSent { request_id, peer } => {
let arrival_time = self.pending_responses_arrival_time.remove(
&(protocol.clone(), request_id).into(),
)
.map(|t| t.elapsed())
.expect(
"Time is added for each inbound request on arrival and only \
removed on success (`ResponseSent`) or failure \
(`InboundFailure`). One can not receive a success event for a \
request that either never arrived, or that has previously \
failed; qed.",
);
let out = Event::InboundRequest {
peer,
protocol: protocol.clone(),
result: Ok(arrival_time),
};
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(out));
}
};
}
}
break Poll::Pending;
}
}
}
#[derive(Debug, derive_more::Display, derive_more::Error)]
pub enum RegisterError {
DuplicateProtocol(#[error(ignore)] Cow<'static, str>),
}
#[derive(Debug, derive_more::Display, derive_more::Error)]
pub enum RequestFailure {
NotConnected,
UnknownProtocol,
Refused,
Obsolete,
#[display(fmt = "Problem on the network")]
Network(#[error(ignore)] OutboundFailure),
}
#[derive(Debug, derive_more::Display, derive_more::Error)]
pub enum ResponseFailure {
#[display(fmt = "Problem on the network")]
Network(#[error(ignore)] InboundFailure),
}
#[derive(Debug, Clone)]
#[doc(hidden)]
pub struct GenericCodec {
max_request_size: u64,
max_response_size: u64,
}
#[async_trait::async_trait]
impl RequestResponseCodec for GenericCodec {
type Protocol = Vec<u8>;
type Request = Vec<u8>;
type Response = Result<Vec<u8>, ()>;
async fn read_request<T>(
&mut self,
_: &Self::Protocol,
mut io: &mut T,
) -> io::Result<Self::Request>
where
T: AsyncRead + Unpin + Send,
{
let length = unsigned_varint::aio::read_usize(&mut io).await
.map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?;
if length > usize::try_from(self.max_request_size).unwrap_or(usize::max_value()) {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("Request size exceeds limit: {} > {}", length, self.max_request_size)
));
}
let mut buffer = vec![0; length];
io.read_exact(&mut buffer).await?;
Ok(buffer)
}
async fn read_response<T>(
&mut self,
_: &Self::Protocol,
mut io: &mut T,
) -> io::Result<Self::Response>
where
T: AsyncRead + Unpin + Send,
{
let length = match unsigned_varint::aio::read_usize(&mut io).await {
Ok(l) => l,
Err(unsigned_varint::io::ReadError::Io(err))
if matches!(err.kind(), io::ErrorKind::UnexpectedEof) =>
{
return Ok(Err(()));
}
Err(err) => return Err(io::Error::new(io::ErrorKind::InvalidInput, err)),
};
if length > usize::try_from(self.max_response_size).unwrap_or(usize::max_value()) {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("Response size exceeds limit: {} > {}", length, self.max_response_size)
));
}
let mut buffer = vec![0; length];
io.read_exact(&mut buffer).await?;
Ok(Ok(buffer))
}
async fn write_request<T>(
&mut self,
_: &Self::Protocol,
io: &mut T,
req: Self::Request,
) -> io::Result<()>
where
T: AsyncWrite + Unpin + Send,
{
{
let mut buffer = unsigned_varint::encode::usize_buffer();
io.write_all(unsigned_varint::encode::usize(req.len(), &mut buffer)).await?;
}
io.write_all(&req).await?;
io.close().await?;
Ok(())
}
async fn write_response<T>(
&mut self,
_: &Self::Protocol,
io: &mut T,
res: Self::Response,
) -> io::Result<()>
where
T: AsyncWrite + Unpin + Send,
{
if let Ok(res) = res {
{
let mut buffer = unsigned_varint::encode::usize_buffer();
io.write_all(unsigned_varint::encode::usize(res.len(), &mut buffer)).await?;
}
io.write_all(&res).await?;
}
io.close().await?;
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use futures::channel::{mpsc, oneshot};
use futures::executor::LocalPool;
use futures::task::Spawn;
use libp2p::identity::Keypair;
use libp2p::Multiaddr;
use libp2p::core::upgrade;
use libp2p::core::transport::{Transport, MemoryTransport};
use libp2p::noise;
use libp2p::swarm::{Swarm, SwarmEvent};
use std::{iter, time::Duration};
fn build_swarm(list: impl Iterator<Item = ProtocolConfig>) -> (Swarm<RequestResponsesBehaviour>, Multiaddr) {
let keypair = Keypair::generate_ed25519();
let noise_keys = noise::Keypair::<noise::X25519Spec>::new()
.into_authentic(&keypair)
.unwrap();
let transport = MemoryTransport
.upgrade(upgrade::Version::V1)
.authenticate(noise::NoiseConfig::xx(noise_keys).into_authenticated())
.multiplex(libp2p::yamux::YamuxConfig::default())
.boxed();
let behaviour = RequestResponsesBehaviour::new(list).unwrap();
let mut swarm = Swarm::new(transport, behaviour, keypair.public().into_peer_id());
let listen_addr: Multiaddr = format!("/memory/{}", rand::random::<u64>()).parse().unwrap();
Swarm::listen_on(&mut swarm, listen_addr.clone()).unwrap();
(swarm, listen_addr)
}
#[test]
fn basic_request_response_works() {
let protocol_name = "/test/req-resp/1";
let mut pool = LocalPool::new();
let mut swarms = (0..2)
.map(|_| {
let (tx, mut rx) = mpsc::channel::<IncomingRequest>(64);
pool.spawner().spawn_obj(async move {
while let Some(rq) = rx.next().await {
assert_eq!(rq.payload, b"this is a request");
let _ = rq.pending_response.send(super::OutgoingResponse {
result: Ok(b"this is a response".to_vec()),
reputation_changes: Vec::new(),
});
}
}.boxed().into()).unwrap();
let protocol_config = ProtocolConfig {
name: From::from(protocol_name),
max_request_size: 1024,
max_response_size: 1024 * 1024,
request_timeout: Duration::from_secs(30),
inbound_queue: Some(tx),
};
build_swarm(iter::once(protocol_config))
})
.collect::<Vec<_>>();
{
let dial_addr = swarms[1].1.clone();
Swarm::dial_addr(&mut swarms[0].0, dial_addr).unwrap();
}
pool.spawner().spawn_obj({
let (mut swarm, _) = swarms.remove(0);
async move {
loop {
match swarm.next_event().await {
SwarmEvent::Behaviour(Event::InboundRequest { result, .. }) => {
result.unwrap();
},
_ => {}
}
}
}.boxed().into()
}).unwrap();
let (mut swarm, _) = swarms.remove(0);
pool.run_until(async move {
let mut response_receiver = None;
loop {
match swarm.next_event().await {
SwarmEvent::ConnectionEstablished { peer_id, .. } => {
let (sender, receiver) = oneshot::channel();
swarm.send_request(
&peer_id,
protocol_name,
b"this is a request".to_vec(),
sender,
IfDisconnected::ImmediateError,
);
assert!(response_receiver.is_none());
response_receiver = Some(receiver);
}
SwarmEvent::Behaviour(Event::RequestFinished {
result, ..
}) => {
result.unwrap();
break;
}
_ => {}
}
}
assert_eq!(response_receiver.unwrap().await.unwrap().unwrap(), b"this is a response");
});
}
#[test]
fn max_response_size_exceeded() {
let protocol_name = "/test/req-resp/1";
let mut pool = LocalPool::new();
let mut swarms = (0..2)
.map(|_| {
let (tx, mut rx) = mpsc::channel::<IncomingRequest>(64);
pool.spawner().spawn_obj(async move {
while let Some(rq) = rx.next().await {
assert_eq!(rq.payload, b"this is a request");
let _ = rq.pending_response.send(super::OutgoingResponse {
result: Ok(b"this response exceeds the limit".to_vec()),
reputation_changes: Vec::new(),
});
}
}.boxed().into()).unwrap();
let protocol_config = ProtocolConfig {
name: From::from(protocol_name),
max_request_size: 1024,
max_response_size: 8,
request_timeout: Duration::from_secs(30),
inbound_queue: Some(tx),
};
build_swarm(iter::once(protocol_config))
})
.collect::<Vec<_>>();
{
let dial_addr = swarms[1].1.clone();
Swarm::dial_addr(&mut swarms[0].0, dial_addr).unwrap();
}
pool.spawner().spawn_obj({
let (mut swarm, _) = swarms.remove(0);
async move {
loop {
match swarm.next_event().await {
SwarmEvent::Behaviour(Event::InboundRequest { result, .. }) => {
assert!(result.is_ok());
break
},
_ => {}
}
}
}.boxed().into()
}).unwrap();
let (mut swarm, _) = swarms.remove(0);
pool.run_until(async move {
let mut response_receiver = None;
loop {
match swarm.next_event().await {
SwarmEvent::ConnectionEstablished { peer_id, .. } => {
let (sender, receiver) = oneshot::channel();
swarm.send_request(
&peer_id,
protocol_name,
b"this is a request".to_vec(),
sender,
IfDisconnected::ImmediateError,
);
assert!(response_receiver.is_none());
response_receiver = Some(receiver);
}
SwarmEvent::Behaviour(Event::RequestFinished {
result, ..
}) => {
assert!(result.is_err());
break;
}
_ => {}
}
}
match response_receiver.unwrap().await.unwrap().unwrap_err() {
RequestFailure::Network(OutboundFailure::ConnectionClosed) => {},
_ => panic!()
}
});
}
#[test]
fn request_id_collision() {
let protocol_name_1 = "/test/req-resp-1/1";
let protocol_name_2 = "/test/req-resp-2/1";
let mut pool = LocalPool::new();
let mut swarm_1 = {
let protocol_configs = vec![
ProtocolConfig {
name: From::from(protocol_name_1),
max_request_size: 1024,
max_response_size: 1024 * 1024,
request_timeout: Duration::from_secs(30),
inbound_queue: None,
},
ProtocolConfig {
name: From::from(protocol_name_2),
max_request_size: 1024,
max_response_size: 1024 * 1024,
request_timeout: Duration::from_secs(30),
inbound_queue: None,
},
];
build_swarm(protocol_configs.into_iter()).0
};
let (mut swarm_2, mut swarm_2_handler_1, mut swarm_2_handler_2, listen_add_2) = {
let (tx_1, rx_1) = mpsc::channel(64);
let (tx_2, rx_2) = mpsc::channel(64);
let protocol_configs = vec![
ProtocolConfig {
name: From::from(protocol_name_1),
max_request_size: 1024,
max_response_size: 1024 * 1024,
request_timeout: Duration::from_secs(30),
inbound_queue: Some(tx_1),
},
ProtocolConfig {
name: From::from(protocol_name_2),
max_request_size: 1024,
max_response_size: 1024 * 1024,
request_timeout: Duration::from_secs(30),
inbound_queue: Some(tx_2),
},
];
let (swarm, listen_addr) = build_swarm(protocol_configs.into_iter());
(swarm, rx_1, rx_2, listen_addr)
};
Swarm::dial_addr(&mut swarm_1, listen_add_2).unwrap();
pool.spawner().spawn_obj(
async move {
loop {
match swarm_2.next_event().await {
SwarmEvent::Behaviour(Event::InboundRequest { result, .. }) => {
result.unwrap();
},
_ => {}
}
}
}.boxed().into()
).unwrap();
pool.spawner().spawn_obj(async move {
let protocol_1_request = swarm_2_handler_1.next().await;
let protocol_2_request = swarm_2_handler_2.next().await;
protocol_1_request.unwrap()
.pending_response
.send(OutgoingResponse {
result: Ok(b"this is a response".to_vec()),
reputation_changes: Vec::new(),
})
.unwrap();
protocol_2_request.unwrap()
.pending_response
.send(OutgoingResponse {
result: Ok(b"this is a response".to_vec()),
reputation_changes: Vec::new(),
})
.unwrap();
}.boxed().into()).unwrap();
pool.run_until(
async move {
let mut response_receivers = None;
let mut num_responses = 0;
loop {
match swarm_1.next_event().await {
SwarmEvent::ConnectionEstablished { peer_id, .. } => {
let (sender_1, receiver_1) = oneshot::channel();
let (sender_2, receiver_2) = oneshot::channel();
swarm_1.send_request(
&peer_id,
protocol_name_1,
b"this is a request".to_vec(),
sender_1,
IfDisconnected::ImmediateError,
);
swarm_1.send_request(
&peer_id,
protocol_name_2,
b"this is a request".to_vec(),
sender_2,
IfDisconnected::ImmediateError,
);
assert!(response_receivers.is_none());
response_receivers = Some((receiver_1, receiver_2));
}
SwarmEvent::Behaviour(Event::RequestFinished {
result, ..
}) => {
num_responses += 1;
result.unwrap();
if num_responses == 2 {
break;
}
}
_ => {}
}
}
let (response_receiver_1, response_receiver_2) = response_receivers.unwrap();
assert_eq!(response_receiver_1.await.unwrap().unwrap(), b"this is a response");
assert_eq!(response_receiver_2.await.unwrap().unwrap(), b"this is a response");
}
);
}
}