use crate::{
PeerId,
nodes::handled_node::{NodeHandler, NodeHandlerEndpoint, NodeHandlerEvent},
nodes::handled_node_tasks::IntoNodeHandler,
protocols_handler::{KeepAlive, ProtocolsHandler, IntoProtocolsHandler, ProtocolsHandlerEvent, ProtocolsHandlerUpgrErr},
upgrade::{
self,
OutboundUpgrade,
InboundUpgradeApply,
OutboundUpgradeApply,
}
};
use futures::prelude::*;
use std::{error, fmt, time::Duration, time::Instant};
use tokio_timer::{Delay, Timeout};
pub struct NodeHandlerWrapperBuilder<TIntoProtoHandler> {
handler: TIntoProtoHandler,
in_timeout: Duration,
out_timeout: Duration,
}
impl<TIntoProtoHandler> NodeHandlerWrapperBuilder<TIntoProtoHandler>
where
TIntoProtoHandler: IntoProtocolsHandler
{
#[inline]
pub(crate) fn new(handler: TIntoProtoHandler, in_timeout: Duration, out_timeout: Duration) -> Self {
NodeHandlerWrapperBuilder {
handler,
in_timeout,
out_timeout,
}
}
#[inline]
pub fn with_in_negotiation_timeout(mut self, timeout: Duration) -> Self {
self.in_timeout = timeout;
self
}
#[inline]
pub fn with_out_negotiation_timeout(mut self, timeout: Duration) -> Self {
self.out_timeout = timeout;
self
}
#[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(),
in_timeout: self.in_timeout,
out_timeout: self.out_timeout,
queued_dial_upgrades: Vec::new(),
unique_dial_upgrade_id: 0,
connection_shutdown: None,
}
}
}
impl<TIntoProtoHandler, TProtoHandler> IntoNodeHandler for NodeHandlerWrapperBuilder<TIntoProtoHandler>
where
TIntoProtoHandler: IntoProtocolsHandler<Handler = TProtoHandler>,
TProtoHandler: ProtocolsHandler,
<TProtoHandler::OutboundProtocol as OutboundUpgrade<<TProtoHandler as ProtocolsHandler>::Substream>>::Error: std::fmt::Debug
{
type Handler = NodeHandlerWrapper<TIntoProtoHandler::Handler>;
fn into_handler(self, remote_peer_id: &PeerId) -> Self::Handler {
NodeHandlerWrapper {
handler: self.handler.into_handler(remote_peer_id),
negotiating_in: Vec::new(),
negotiating_out: Vec::new(),
in_timeout: self.in_timeout,
out_timeout: self.out_timeout,
queued_dial_upgrades: Vec::new(),
unique_dial_upgrade_id: 0,
connection_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>>,
)>,
in_timeout: Duration,
out_timeout: Duration,
queued_dial_upgrades: Vec<(u64, TProtoHandler::OutboundProtocol)>,
unique_dial_upgrade_id: u64,
connection_shutdown: Option<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,
<TProtoHandler::OutboundProtocol as OutboundUpgrade<<TProtoHandler as ProtocolsHandler>::Substream>>::Error: std::fmt::Debug
{
type InEvent = TProtoHandler::InEvent;
type OutEvent = TProtoHandler::OutEvent;
type Error = NodeHandlerWrapperError<TProtoHandler::Error>;
type Substream = TProtoHandler::Substream;
type OutboundOpenInfo = (u64, TProtoHandler::OutboundOpenInfo);
fn inject_substream(
&mut self,
substream: Self::Substream,
endpoint: NodeHandlerEndpoint<Self::OutboundOpenInfo>,
) {
match endpoint {
NodeHandlerEndpoint::Listener => {
let protocol = self.handler.listen_protocol();
let upgrade = upgrade::apply_inbound(substream, protocol);
let with_timeout = Timeout::new(upgrade, self.in_timeout);
self.negotiating_in.push(with_timeout);
}
NodeHandlerEndpoint::Dialer((upgrade_id, user_data)) => {
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, self.out_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_inner and is_elapsed 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()?;
self.connection_shutdown = match self.handler.connection_keep_alive() {
KeepAlive::Until(expiration) => Some(Delay::new(expiration)),
KeepAlive::Now => Some(Delay::new(Instant::now())),
KeepAlive::Forever => None,
};
match poll_result {
Async::Ready(ProtocolsHandlerEvent::Custom(event)) => {
return Ok(Async::Ready(NodeHandlerEvent::Custom(event)));
}
Async::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
upgrade,
info,
}) => {
let id = self.unique_dial_upgrade_id;
self.unique_dial_upgrade_id += 1;
self.queued_dial_upgrades.push((id, upgrade));
return Ok(Async::Ready(
NodeHandlerEvent::OutboundSubstreamRequest((id, info)),
));
}
Async::NotReady => (),
};
if let Some(mut connection_shutdown) = self.connection_shutdown.take() {
if self.negotiating_in.is_empty() && self.negotiating_out.is_empty() {
match connection_shutdown.poll() {
Ok(Async::Ready(_)) | Err(_) => {
return Err(NodeHandlerWrapperError::UselessTimeout);
},
Ok(Async::NotReady) => {
self.connection_shutdown = Some(connection_shutdown);
}
}
} else {
self.connection_shutdown = Some(connection_shutdown);
}
}
Ok(Async::NotReady)
}
}