use crate::protocols_handler::{
KeepAlive,
ProtocolsHandler,
IntoProtocolsHandler,
ProtocolsHandlerEvent,
ProtocolsHandlerUpgrErr
};
use futures::prelude::*;
use libp2p_core::{
ConnectedPoint,
PeerId,
nodes::collection::ConnectionInfo,
nodes::handled_node::{IntoNodeHandler, NodeHandler, NodeHandlerEndpoint, NodeHandlerEvent},
upgrade::{self, InboundUpgradeApply, OutboundUpgradeApply}
};
use std::{error, fmt, time::Duration};
use wasm_timer::{Delay, Timeout};
pub struct NodeHandlerWrapperBuilder<TIntoProtoHandler> {
handler: TIntoProtoHandler,
}
impl<TIntoProtoHandler> NodeHandlerWrapperBuilder<TIntoProtoHandler>
where
TIntoProtoHandler: IntoProtocolsHandler
{
#[inline]
pub(crate) fn new(handler: TIntoProtoHandler) -> Self {
NodeHandlerWrapperBuilder {
handler,
}
}
#[deprecated(note = "Pass the NodeHandlerWrapperBuilder directly")]
#[inline]
pub fn build(self) -> NodeHandlerWrapper<TIntoProtoHandler>
where TIntoProtoHandler: ProtocolsHandler
{
NodeHandlerWrapper {
handler: self.handler,
negotiating_in: Vec::new(),
negotiating_out: Vec::new(),
queued_dial_upgrades: Vec::new(),
unique_dial_upgrade_id: 0,
shutdown: Shutdown::None,
}
}
}
impl<TIntoProtoHandler, TProtoHandler, TConnInfo> IntoNodeHandler<(TConnInfo, ConnectedPoint)>
for NodeHandlerWrapperBuilder<TIntoProtoHandler>
where
TIntoProtoHandler: IntoProtocolsHandler<Handler = TProtoHandler>,
TProtoHandler: ProtocolsHandler,
TConnInfo: ConnectionInfo<PeerId = PeerId>,
{
type Handler = NodeHandlerWrapper<TIntoProtoHandler::Handler>;
fn into_handler(self, remote_info: &(TConnInfo, ConnectedPoint)) -> Self::Handler {
NodeHandlerWrapper {
handler: self.handler.into_handler(&remote_info.0.peer_id(), &remote_info.1),
negotiating_in: Vec::new(),
negotiating_out: Vec::new(),
queued_dial_upgrades: Vec::new(),
unique_dial_upgrade_id: 0,
shutdown: Shutdown::None,
}
}
}
pub struct NodeHandlerWrapper<TProtoHandler>
where
TProtoHandler: ProtocolsHandler,
{
handler: TProtoHandler,
negotiating_in:
Vec<Timeout<InboundUpgradeApply<TProtoHandler::Substream, TProtoHandler::InboundProtocol>>>,
negotiating_out: Vec<(
TProtoHandler::OutboundOpenInfo,
Timeout<OutboundUpgradeApply<TProtoHandler::Substream, TProtoHandler::OutboundProtocol>>,
)>,
queued_dial_upgrades: Vec<(u64, TProtoHandler::OutboundProtocol)>,
unique_dial_upgrade_id: u64,
shutdown: Shutdown,
}
enum Shutdown {
None,
Asap,
Later(Delay)
}
#[derive(Debug)]
pub enum NodeHandlerWrapperError<TErr> {
Handler(TErr),
UselessTimeout,
}
impl<TErr> From<TErr> for NodeHandlerWrapperError<TErr> {
fn from(err: TErr) -> NodeHandlerWrapperError<TErr> {
NodeHandlerWrapperError::Handler(err)
}
}
impl<TErr> fmt::Display for NodeHandlerWrapperError<TErr>
where
TErr: fmt::Display
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
NodeHandlerWrapperError::Handler(err) => write!(f, "{}", err),
NodeHandlerWrapperError::UselessTimeout =>
write!(f, "Node has been closed due to inactivity"),
}
}
}
impl<TErr> error::Error for NodeHandlerWrapperError<TErr>
where
TErr: error::Error + 'static
{
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
match self {
NodeHandlerWrapperError::Handler(err) => Some(err),
NodeHandlerWrapperError::UselessTimeout => None,
}
}
}
impl<TProtoHandler> NodeHandler for NodeHandlerWrapper<TProtoHandler>
where
TProtoHandler: ProtocolsHandler,
{
type InEvent = TProtoHandler::InEvent;
type OutEvent = TProtoHandler::OutEvent;
type Error = NodeHandlerWrapperError<TProtoHandler::Error>;
type Substream = TProtoHandler::Substream;
type OutboundOpenInfo = (u64, TProtoHandler::OutboundOpenInfo, Duration);
fn inject_substream(
&mut self,
substream: Self::Substream,
endpoint: NodeHandlerEndpoint<Self::OutboundOpenInfo>,
) {
match endpoint {
NodeHandlerEndpoint::Listener => {
let protocol = self.handler.listen_protocol();
let timeout = protocol.timeout().clone();
let upgrade = upgrade::apply_inbound(substream, protocol.into_upgrade());
let with_timeout = Timeout::new(upgrade, timeout);
self.negotiating_in.push(with_timeout);
}
NodeHandlerEndpoint::Dialer((upgrade_id, user_data, timeout)) => {
let pos = match self
.queued_dial_upgrades
.iter()
.position(|(id, _)| id == &upgrade_id)
{
Some(p) => p,
None => {
debug_assert!(false, "Received an upgrade with an invalid upgrade ID");
return;
}
};
let (_, proto_upgrade) = self.queued_dial_upgrades.remove(pos);
let upgrade = upgrade::apply_outbound(substream, proto_upgrade);
let with_timeout = Timeout::new(upgrade, timeout);
self.negotiating_out.push((user_data, with_timeout));
}
}
}
#[inline]
fn inject_event(&mut self, event: Self::InEvent) {
self.handler.inject_event(event);
}
fn poll(&mut self) -> Poll<NodeHandlerEvent<Self::OutboundOpenInfo, Self::OutEvent>, Self::Error> {
for n in (0..self.negotiating_in.len()).rev() {
let mut in_progress = self.negotiating_in.swap_remove(n);
match in_progress.poll() {
Ok(Async::Ready(upgrade)) =>
self.handler.inject_fully_negotiated_inbound(upgrade),
Ok(Async::NotReady) => self.negotiating_in.push(in_progress),
Err(_err) => {}
}
}
for n in (0..self.negotiating_out.len()).rev() {
let (upgr_info, mut in_progress) = self.negotiating_out.swap_remove(n);
match in_progress.poll() {
Ok(Async::Ready(upgrade)) => {
self.handler.inject_fully_negotiated_outbound(upgrade, upgr_info);
}
Ok(Async::NotReady) => {
self.negotiating_out.push((upgr_info, in_progress));
}
Err(err) => {
let err = if err.is_elapsed() {
ProtocolsHandlerUpgrErr::Timeout
} else if err.is_timer() {
ProtocolsHandlerUpgrErr::Timer
} else {
debug_assert!(err.is_inner());
let err = err.into_inner().expect("Timeout error is one of {elapsed, \
timer, inner}; is_elapsed and is_timer are both false; error is \
inner; QED");
ProtocolsHandlerUpgrErr::Upgrade(err)
};
self.handler.inject_dial_upgrade_error(upgr_info, err);
}
}
}
let poll_result = self.handler.poll()?;
match (&mut self.shutdown, self.handler.connection_keep_alive()) {
(Shutdown::Later(d), KeepAlive::Until(t)) =>
if d.deadline() != t {
d.reset(t)
},
(_, KeepAlive::Until(t)) => self.shutdown = Shutdown::Later(Delay::new(t)),
(_, KeepAlive::No) => self.shutdown = Shutdown::Asap,
(_, KeepAlive::Yes) => self.shutdown = Shutdown::None
};
match poll_result {
Async::Ready(ProtocolsHandlerEvent::Custom(event)) => {
return Ok(Async::Ready(NodeHandlerEvent::Custom(event)));
}
Async::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol,
info,
}) => {
let id = self.unique_dial_upgrade_id;
let timeout = protocol.timeout().clone();
self.unique_dial_upgrade_id += 1;
self.queued_dial_upgrades.push((id, protocol.into_upgrade()));
return Ok(Async::Ready(
NodeHandlerEvent::OutboundSubstreamRequest((id, info, timeout)),
));
}
Async::NotReady => (),
};
if self.negotiating_in.is_empty() && self.negotiating_out.is_empty() {
match self.shutdown {
Shutdown::None => {},
Shutdown::Asap => return Err(NodeHandlerWrapperError::UselessTimeout),
Shutdown::Later(ref mut delay) => match delay.poll() {
Ok(Async::Ready(_)) | Err(_) =>
return Err(NodeHandlerWrapperError::UselessTimeout),
Ok(Async::NotReady) => {}
}
}
}
Ok(Async::NotReady)
}
}