use crate::protocol::{
KadInStreamSink, KadOutStreamSink, KadPeer, KadRequestMsg, KadResponseMsg,
KademliaProtocolConfig,
};
use crate::record::{self, Record};
use futures::prelude::*;
use futures::stream::SelectAll;
use instant::Instant;
use libp2p_core::{either::EitherOutput, upgrade, ConnectedPoint, PeerId};
use libp2p_swarm::handler::{
ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
};
use libp2p_swarm::{
ConnectionHandler, ConnectionHandlerEvent, ConnectionHandlerUpgrErr, IntoConnectionHandler,
KeepAlive, NegotiatedSubstream, SubstreamProtocol,
};
use log::trace;
use std::task::Waker;
use std::{
error, fmt, io, marker::PhantomData, pin::Pin, task::Context, task::Poll, time::Duration,
};
const MAX_NUM_INBOUND_SUBSTREAMS: usize = 32;
pub struct KademliaHandlerProto<T> {
config: KademliaHandlerConfig,
_type: PhantomData<T>,
}
impl<T> KademliaHandlerProto<T> {
pub fn new(config: KademliaHandlerConfig) -> Self {
KademliaHandlerProto {
config,
_type: PhantomData,
}
}
}
impl<T: Clone + fmt::Debug + Send + 'static + Unpin> IntoConnectionHandler
for KademliaHandlerProto<T>
{
type Handler = KademliaHandler<T>;
fn into_handler(self, remote_peer_id: &PeerId, endpoint: &ConnectedPoint) -> Self::Handler {
KademliaHandler::new(self.config, endpoint.clone(), *remote_peer_id)
}
fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol {
if self.config.allow_listening {
upgrade::EitherUpgrade::A(self.config.protocol_config.clone())
} else {
upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade)
}
}
}
pub struct KademliaHandler<TUserData> {
config: KademliaHandlerConfig,
next_connec_unique_id: UniqueConnecId,
outbound_substreams: SelectAll<OutboundSubstreamState<TUserData>>,
inbound_substreams: SelectAll<InboundSubstreamState<TUserData>>,
keep_alive: KeepAlive,
endpoint: ConnectedPoint,
remote_peer_id: PeerId,
protocol_status: ProtocolStatus,
}
enum ProtocolStatus {
Unconfirmed,
Confirmed,
Reported,
}
#[derive(Debug, Clone)]
pub struct KademliaHandlerConfig {
pub protocol_config: KademliaProtocolConfig,
pub allow_listening: bool,
pub idle_timeout: Duration,
}
enum OutboundSubstreamState<TUserData> {
PendingOpen(SubstreamProtocol<KademliaProtocolConfig, (KadRequestMsg, Option<TUserData>)>),
PendingSend(
KadOutStreamSink<NegotiatedSubstream>,
KadRequestMsg,
Option<TUserData>,
),
PendingFlush(KadOutStreamSink<NegotiatedSubstream>, Option<TUserData>),
WaitingAnswer(KadOutStreamSink<NegotiatedSubstream>, TUserData),
ReportError(KademliaHandlerQueryErr, TUserData),
Closing(KadOutStreamSink<NegotiatedSubstream>),
Done,
Poisoned,
}
enum InboundSubstreamState<TUserData> {
WaitingMessage {
first: bool,
connection_id: UniqueConnecId,
substream: KadInStreamSink<NegotiatedSubstream>,
},
WaitingBehaviour(
UniqueConnecId,
KadInStreamSink<NegotiatedSubstream>,
Option<Waker>,
),
PendingSend(
UniqueConnecId,
KadInStreamSink<NegotiatedSubstream>,
KadResponseMsg,
),
PendingFlush(UniqueConnecId, KadInStreamSink<NegotiatedSubstream>),
Closing(KadInStreamSink<NegotiatedSubstream>),
Cancelled,
Poisoned {
phantom: PhantomData<TUserData>,
},
}
impl<TUserData> InboundSubstreamState<TUserData> {
fn try_answer_with(
&mut self,
id: KademliaRequestId,
msg: KadResponseMsg,
) -> Result<(), KadResponseMsg> {
match std::mem::replace(
self,
InboundSubstreamState::Poisoned {
phantom: PhantomData,
},
) {
InboundSubstreamState::WaitingBehaviour(conn_id, substream, mut waker)
if conn_id == id.connec_unique_id =>
{
*self = InboundSubstreamState::PendingSend(conn_id, substream, msg);
if let Some(waker) = waker.take() {
waker.wake();
}
Ok(())
}
other => {
*self = other;
Err(msg)
}
}
}
fn close(&mut self) {
match std::mem::replace(
self,
InboundSubstreamState::Poisoned {
phantom: PhantomData,
},
) {
InboundSubstreamState::WaitingMessage { substream, .. }
| InboundSubstreamState::WaitingBehaviour(_, substream, _)
| InboundSubstreamState::PendingSend(_, substream, _)
| InboundSubstreamState::PendingFlush(_, substream)
| InboundSubstreamState::Closing(substream) => {
*self = InboundSubstreamState::Closing(substream);
}
InboundSubstreamState::Cancelled => {
*self = InboundSubstreamState::Cancelled;
}
InboundSubstreamState::Poisoned { .. } => unreachable!(),
}
}
}
#[derive(Debug)]
pub enum KademliaHandlerEvent<TUserData> {
ProtocolConfirmed { endpoint: ConnectedPoint },
FindNodeReq {
key: Vec<u8>,
request_id: KademliaRequestId,
},
FindNodeRes {
closer_peers: Vec<KadPeer>,
user_data: TUserData,
},
GetProvidersReq {
key: record::Key,
request_id: KademliaRequestId,
},
GetProvidersRes {
closer_peers: Vec<KadPeer>,
provider_peers: Vec<KadPeer>,
user_data: TUserData,
},
QueryError {
error: KademliaHandlerQueryErr,
user_data: TUserData,
},
AddProvider {
key: record::Key,
provider: KadPeer,
},
GetRecord {
key: record::Key,
request_id: KademliaRequestId,
},
GetRecordRes {
record: Option<Record>,
closer_peers: Vec<KadPeer>,
user_data: TUserData,
},
PutRecord {
record: Record,
request_id: KademliaRequestId,
},
PutRecordRes {
key: record::Key,
value: Vec<u8>,
user_data: TUserData,
},
}
#[derive(Debug)]
pub enum KademliaHandlerQueryErr {
Upgrade(ConnectionHandlerUpgrErr<io::Error>),
UnexpectedMessage,
Io(io::Error),
}
impl fmt::Display for KademliaHandlerQueryErr {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
KademliaHandlerQueryErr::Upgrade(err) => {
write!(f, "Error while performing Kademlia query: {}", err)
}
KademliaHandlerQueryErr::UnexpectedMessage => {
write!(
f,
"Remote answered our Kademlia RPC query with the wrong message type"
)
}
KademliaHandlerQueryErr::Io(err) => {
write!(f, "I/O error during a Kademlia RPC query: {}", err)
}
}
}
}
impl error::Error for KademliaHandlerQueryErr {
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
match self {
KademliaHandlerQueryErr::Upgrade(err) => Some(err),
KademliaHandlerQueryErr::UnexpectedMessage => None,
KademliaHandlerQueryErr::Io(err) => Some(err),
}
}
}
impl From<ConnectionHandlerUpgrErr<io::Error>> for KademliaHandlerQueryErr {
fn from(err: ConnectionHandlerUpgrErr<io::Error>) -> Self {
KademliaHandlerQueryErr::Upgrade(err)
}
}
#[derive(Debug)]
pub enum KademliaHandlerIn<TUserData> {
Reset(KademliaRequestId),
FindNodeReq {
key: Vec<u8>,
user_data: TUserData,
},
FindNodeRes {
closer_peers: Vec<KadPeer>,
request_id: KademliaRequestId,
},
GetProvidersReq {
key: record::Key,
user_data: TUserData,
},
GetProvidersRes {
closer_peers: Vec<KadPeer>,
provider_peers: Vec<KadPeer>,
request_id: KademliaRequestId,
},
AddProvider {
key: record::Key,
provider: KadPeer,
},
GetRecord {
key: record::Key,
user_data: TUserData,
},
GetRecordRes {
record: Option<Record>,
closer_peers: Vec<KadPeer>,
request_id: KademliaRequestId,
},
PutRecord {
record: Record,
user_data: TUserData,
},
PutRecordRes {
key: record::Key,
value: Vec<u8>,
request_id: KademliaRequestId,
},
}
#[derive(Debug, PartialEq, Eq, Copy, Clone)]
pub struct KademliaRequestId {
connec_unique_id: UniqueConnecId,
}
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
struct UniqueConnecId(u64);
impl<TUserData> KademliaHandler<TUserData>
where
TUserData: Clone + fmt::Debug + Send + 'static + Unpin,
{
pub fn new(
config: KademliaHandlerConfig,
endpoint: ConnectedPoint,
remote_peer_id: PeerId,
) -> Self {
let keep_alive = KeepAlive::Until(Instant::now() + config.idle_timeout);
KademliaHandler {
config,
endpoint,
remote_peer_id,
next_connec_unique_id: UniqueConnecId(0),
inbound_substreams: Default::default(),
outbound_substreams: Default::default(),
keep_alive,
protocol_status: ProtocolStatus::Unconfirmed,
}
}
fn on_fully_negotiated_outbound(
&mut self,
FullyNegotiatedOutbound {
protocol,
info: (msg, user_data),
}: FullyNegotiatedOutbound<
<Self as ConnectionHandler>::OutboundProtocol,
<Self as ConnectionHandler>::OutboundOpenInfo,
>,
) {
self.outbound_substreams
.push(OutboundSubstreamState::PendingSend(
protocol, msg, user_data,
));
if let ProtocolStatus::Unconfirmed = self.protocol_status {
self.protocol_status = ProtocolStatus::Confirmed;
}
}
fn on_fully_negotiated_inbound(
&mut self,
FullyNegotiatedInbound { protocol, .. }: FullyNegotiatedInbound<
<Self as ConnectionHandler>::InboundProtocol,
<Self as ConnectionHandler>::InboundOpenInfo,
>,
) {
let protocol = match protocol {
EitherOutput::First(p) => p,
EitherOutput::Second(p) => void::unreachable(p),
};
if let ProtocolStatus::Unconfirmed = self.protocol_status {
self.protocol_status = ProtocolStatus::Confirmed;
}
if self.inbound_substreams.len() == MAX_NUM_INBOUND_SUBSTREAMS {
if let Some(s) = self.inbound_substreams.iter_mut().find(|s| {
matches!(
s,
InboundSubstreamState::WaitingMessage { first: false, .. }
)
}) {
*s = InboundSubstreamState::Cancelled;
log::warn!(
"New inbound substream to {:?} exceeds inbound substream limit. \
Removed older substream waiting to be reused.",
self.remote_peer_id,
)
} else {
log::warn!(
"New inbound substream to {:?} exceeds inbound substream limit. \
No older substream waiting to be reused. Dropping new substream.",
self.remote_peer_id,
);
return;
}
}
debug_assert!(self.config.allow_listening);
let connec_unique_id = self.next_connec_unique_id;
self.next_connec_unique_id.0 += 1;
self.inbound_substreams
.push(InboundSubstreamState::WaitingMessage {
first: true,
connection_id: connec_unique_id,
substream: protocol,
});
}
fn on_dial_upgrade_error(
&mut self,
DialUpgradeError {
info: (_, user_data),
error,
..
}: DialUpgradeError<
<Self as ConnectionHandler>::OutboundOpenInfo,
<Self as ConnectionHandler>::OutboundProtocol,
>,
) {
if let Some(user_data) = user_data {
self.outbound_substreams
.push(OutboundSubstreamState::ReportError(error.into(), user_data));
}
}
}
impl<TUserData> ConnectionHandler for KademliaHandler<TUserData>
where
TUserData: Clone + fmt::Debug + Send + 'static + Unpin,
{
type InEvent = KademliaHandlerIn<TUserData>;
type OutEvent = KademliaHandlerEvent<TUserData>;
type Error = io::Error; type InboundProtocol = upgrade::EitherUpgrade<KademliaProtocolConfig, upgrade::DeniedUpgrade>;
type OutboundProtocol = KademliaProtocolConfig;
type OutboundOpenInfo = (KadRequestMsg, Option<TUserData>);
type InboundOpenInfo = ();
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
if self.config.allow_listening {
SubstreamProtocol::new(self.config.protocol_config.clone(), ())
.map_upgrade(upgrade::EitherUpgrade::A)
} else {
SubstreamProtocol::new(upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade), ())
}
}
fn on_behaviour_event(&mut self, message: KademliaHandlerIn<TUserData>) {
match message {
KademliaHandlerIn::Reset(request_id) => {
if let Some(state) = self
.inbound_substreams
.iter_mut()
.find(|state| match state {
InboundSubstreamState::WaitingBehaviour(conn_id, _, _) => {
conn_id == &request_id.connec_unique_id
}
_ => false,
})
{
state.close();
}
}
KademliaHandlerIn::FindNodeReq { key, user_data } => {
let msg = KadRequestMsg::FindNode { key };
self.outbound_substreams
.push(OutboundSubstreamState::PendingOpen(SubstreamProtocol::new(
self.config.protocol_config.clone(),
(msg, Some(user_data)),
)));
}
KademliaHandlerIn::FindNodeRes {
closer_peers,
request_id,
} => self.answer_pending_request(request_id, KadResponseMsg::FindNode { closer_peers }),
KademliaHandlerIn::GetProvidersReq { key, user_data } => {
let msg = KadRequestMsg::GetProviders { key };
self.outbound_substreams
.push(OutboundSubstreamState::PendingOpen(SubstreamProtocol::new(
self.config.protocol_config.clone(),
(msg, Some(user_data)),
)));
}
KademliaHandlerIn::GetProvidersRes {
closer_peers,
provider_peers,
request_id,
} => self.answer_pending_request(
request_id,
KadResponseMsg::GetProviders {
closer_peers,
provider_peers,
},
),
KademliaHandlerIn::AddProvider { key, provider } => {
let msg = KadRequestMsg::AddProvider { key, provider };
self.outbound_substreams
.push(OutboundSubstreamState::PendingOpen(SubstreamProtocol::new(
self.config.protocol_config.clone(),
(msg, None),
)));
}
KademliaHandlerIn::GetRecord { key, user_data } => {
let msg = KadRequestMsg::GetValue { key };
self.outbound_substreams
.push(OutboundSubstreamState::PendingOpen(SubstreamProtocol::new(
self.config.protocol_config.clone(),
(msg, Some(user_data)),
)));
}
KademliaHandlerIn::PutRecord { record, user_data } => {
let msg = KadRequestMsg::PutValue { record };
self.outbound_substreams
.push(OutboundSubstreamState::PendingOpen(SubstreamProtocol::new(
self.config.protocol_config.clone(),
(msg, Some(user_data)),
)));
}
KademliaHandlerIn::GetRecordRes {
record,
closer_peers,
request_id,
} => {
self.answer_pending_request(
request_id,
KadResponseMsg::GetValue {
record,
closer_peers,
},
);
}
KademliaHandlerIn::PutRecordRes {
key,
request_id,
value,
} => {
self.answer_pending_request(request_id, KadResponseMsg::PutValue { key, value });
}
}
}
fn connection_keep_alive(&self) -> KeepAlive {
self.keep_alive
}
fn poll(
&mut self,
cx: &mut Context<'_>,
) -> Poll<
ConnectionHandlerEvent<
Self::OutboundProtocol,
Self::OutboundOpenInfo,
Self::OutEvent,
Self::Error,
>,
> {
if let ProtocolStatus::Confirmed = self.protocol_status {
self.protocol_status = ProtocolStatus::Reported;
return Poll::Ready(ConnectionHandlerEvent::Custom(
KademliaHandlerEvent::ProtocolConfirmed {
endpoint: self.endpoint.clone(),
},
));
}
if let Poll::Ready(Some(event)) = self.outbound_substreams.poll_next_unpin(cx) {
return Poll::Ready(event);
}
if let Poll::Ready(Some(event)) = self.inbound_substreams.poll_next_unpin(cx) {
return Poll::Ready(event);
}
if self.outbound_substreams.is_empty() && self.inbound_substreams.is_empty() {
self.keep_alive = KeepAlive::Until(Instant::now() + self.config.idle_timeout);
} else {
self.keep_alive = KeepAlive::Yes;
}
Poll::Pending
}
fn on_connection_event(
&mut self,
event: ConnectionEvent<
Self::InboundProtocol,
Self::OutboundProtocol,
Self::InboundOpenInfo,
Self::OutboundOpenInfo,
>,
) {
match event {
ConnectionEvent::FullyNegotiatedOutbound(fully_negotiated_outbound) => {
self.on_fully_negotiated_outbound(fully_negotiated_outbound)
}
ConnectionEvent::FullyNegotiatedInbound(fully_negotiated_inbound) => {
self.on_fully_negotiated_inbound(fully_negotiated_inbound)
}
ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {
self.on_dial_upgrade_error(dial_upgrade_error)
}
ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) => {}
}
}
}
impl<TUserData> KademliaHandler<TUserData>
where
TUserData: 'static + Clone + Send + Unpin + fmt::Debug,
{
fn answer_pending_request(&mut self, request_id: KademliaRequestId, mut msg: KadResponseMsg) {
for state in self.inbound_substreams.iter_mut() {
match state.try_answer_with(request_id, msg) {
Ok(()) => return,
Err(m) => {
msg = m;
}
}
}
debug_assert!(false, "Cannot find inbound substream for {request_id:?}")
}
}
impl Default for KademliaHandlerConfig {
fn default() -> Self {
KademliaHandlerConfig {
protocol_config: Default::default(),
allow_listening: true,
idle_timeout: Duration::from_secs(10),
}
}
}
impl<TUserData> Stream for OutboundSubstreamState<TUserData>
where
TUserData: Unpin,
{
type Item = ConnectionHandlerEvent<
KademliaProtocolConfig,
(KadRequestMsg, Option<TUserData>),
KademliaHandlerEvent<TUserData>,
io::Error,
>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = self.get_mut();
loop {
match std::mem::replace(this, OutboundSubstreamState::Poisoned) {
OutboundSubstreamState::PendingOpen(protocol) => {
*this = OutboundSubstreamState::Done;
return Poll::Ready(Some(ConnectionHandlerEvent::OutboundSubstreamRequest {
protocol,
}));
}
OutboundSubstreamState::PendingSend(mut substream, msg, user_data) => {
match substream.poll_ready_unpin(cx) {
Poll::Ready(Ok(())) => match substream.start_send_unpin(msg) {
Ok(()) => {
*this = OutboundSubstreamState::PendingFlush(substream, user_data);
}
Err(error) => {
*this = OutboundSubstreamState::Done;
let event = user_data.map(|user_data| {
ConnectionHandlerEvent::Custom(
KademliaHandlerEvent::QueryError {
error: KademliaHandlerQueryErr::Io(error),
user_data,
},
)
});
return Poll::Ready(event);
}
},
Poll::Pending => {
*this = OutboundSubstreamState::PendingSend(substream, msg, user_data);
return Poll::Pending;
}
Poll::Ready(Err(error)) => {
*this = OutboundSubstreamState::Done;
let event = user_data.map(|user_data| {
ConnectionHandlerEvent::Custom(KademliaHandlerEvent::QueryError {
error: KademliaHandlerQueryErr::Io(error),
user_data,
})
});
return Poll::Ready(event);
}
}
}
OutboundSubstreamState::PendingFlush(mut substream, user_data) => {
match substream.poll_flush_unpin(cx) {
Poll::Ready(Ok(())) => {
if let Some(user_data) = user_data {
*this = OutboundSubstreamState::WaitingAnswer(substream, user_data);
} else {
*this = OutboundSubstreamState::Closing(substream);
}
}
Poll::Pending => {
*this = OutboundSubstreamState::PendingFlush(substream, user_data);
return Poll::Pending;
}
Poll::Ready(Err(error)) => {
*this = OutboundSubstreamState::Done;
let event = user_data.map(|user_data| {
ConnectionHandlerEvent::Custom(KademliaHandlerEvent::QueryError {
error: KademliaHandlerQueryErr::Io(error),
user_data,
})
});
return Poll::Ready(event);
}
}
}
OutboundSubstreamState::WaitingAnswer(mut substream, user_data) => {
match substream.poll_next_unpin(cx) {
Poll::Ready(Some(Ok(msg))) => {
*this = OutboundSubstreamState::Closing(substream);
let event = process_kad_response(msg, user_data);
return Poll::Ready(Some(ConnectionHandlerEvent::Custom(event)));
}
Poll::Pending => {
*this = OutboundSubstreamState::WaitingAnswer(substream, user_data);
return Poll::Pending;
}
Poll::Ready(Some(Err(error))) => {
*this = OutboundSubstreamState::Done;
let event = KademliaHandlerEvent::QueryError {
error: KademliaHandlerQueryErr::Io(error),
user_data,
};
return Poll::Ready(Some(ConnectionHandlerEvent::Custom(event)));
}
Poll::Ready(None) => {
*this = OutboundSubstreamState::Done;
let event = KademliaHandlerEvent::QueryError {
error: KademliaHandlerQueryErr::Io(
io::ErrorKind::UnexpectedEof.into(),
),
user_data,
};
return Poll::Ready(Some(ConnectionHandlerEvent::Custom(event)));
}
}
}
OutboundSubstreamState::ReportError(error, user_data) => {
*this = OutboundSubstreamState::Done;
let event = KademliaHandlerEvent::QueryError { error, user_data };
return Poll::Ready(Some(ConnectionHandlerEvent::Custom(event)));
}
OutboundSubstreamState::Closing(mut stream) => match stream.poll_close_unpin(cx) {
Poll::Ready(Ok(())) | Poll::Ready(Err(_)) => return Poll::Ready(None),
Poll::Pending => {
*this = OutboundSubstreamState::Closing(stream);
return Poll::Pending;
}
},
OutboundSubstreamState::Done => {
*this = OutboundSubstreamState::Done;
return Poll::Ready(None);
}
OutboundSubstreamState::Poisoned => unreachable!(),
}
}
}
}
impl<TUserData> Stream for InboundSubstreamState<TUserData>
where
TUserData: Unpin,
{
type Item = ConnectionHandlerEvent<
KademliaProtocolConfig,
(KadRequestMsg, Option<TUserData>),
KademliaHandlerEvent<TUserData>,
io::Error,
>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = self.get_mut();
loop {
match std::mem::replace(
this,
Self::Poisoned {
phantom: PhantomData,
},
) {
InboundSubstreamState::WaitingMessage {
first,
connection_id,
mut substream,
} => match substream.poll_next_unpin(cx) {
Poll::Ready(Some(Ok(KadRequestMsg::Ping))) => {
log::warn!("Kademlia PING messages are unsupported");
*this = InboundSubstreamState::Closing(substream);
}
Poll::Ready(Some(Ok(KadRequestMsg::FindNode { key }))) => {
*this =
InboundSubstreamState::WaitingBehaviour(connection_id, substream, None);
return Poll::Ready(Some(ConnectionHandlerEvent::Custom(
KademliaHandlerEvent::FindNodeReq {
key,
request_id: KademliaRequestId {
connec_unique_id: connection_id,
},
},
)));
}
Poll::Ready(Some(Ok(KadRequestMsg::GetProviders { key }))) => {
*this =
InboundSubstreamState::WaitingBehaviour(connection_id, substream, None);
return Poll::Ready(Some(ConnectionHandlerEvent::Custom(
KademliaHandlerEvent::GetProvidersReq {
key,
request_id: KademliaRequestId {
connec_unique_id: connection_id,
},
},
)));
}
Poll::Ready(Some(Ok(KadRequestMsg::AddProvider { key, provider }))) => {
*this = InboundSubstreamState::WaitingMessage {
first: false,
connection_id,
substream,
};
return Poll::Ready(Some(ConnectionHandlerEvent::Custom(
KademliaHandlerEvent::AddProvider { key, provider },
)));
}
Poll::Ready(Some(Ok(KadRequestMsg::GetValue { key }))) => {
*this =
InboundSubstreamState::WaitingBehaviour(connection_id, substream, None);
return Poll::Ready(Some(ConnectionHandlerEvent::Custom(
KademliaHandlerEvent::GetRecord {
key,
request_id: KademliaRequestId {
connec_unique_id: connection_id,
},
},
)));
}
Poll::Ready(Some(Ok(KadRequestMsg::PutValue { record }))) => {
*this =
InboundSubstreamState::WaitingBehaviour(connection_id, substream, None);
return Poll::Ready(Some(ConnectionHandlerEvent::Custom(
KademliaHandlerEvent::PutRecord {
record,
request_id: KademliaRequestId {
connec_unique_id: connection_id,
},
},
)));
}
Poll::Pending => {
*this = InboundSubstreamState::WaitingMessage {
first,
connection_id,
substream,
};
return Poll::Pending;
}
Poll::Ready(None) => {
return Poll::Ready(None);
}
Poll::Ready(Some(Err(e))) => {
trace!("Inbound substream error: {:?}", e);
return Poll::Ready(None);
}
},
InboundSubstreamState::WaitingBehaviour(id, substream, _) => {
*this = InboundSubstreamState::WaitingBehaviour(
id,
substream,
Some(cx.waker().clone()),
);
return Poll::Pending;
}
InboundSubstreamState::PendingSend(id, mut substream, msg) => {
match substream.poll_ready_unpin(cx) {
Poll::Ready(Ok(())) => match substream.start_send_unpin(msg) {
Ok(()) => {
*this = InboundSubstreamState::PendingFlush(id, substream);
}
Err(_) => return Poll::Ready(None),
},
Poll::Pending => {
*this = InboundSubstreamState::PendingSend(id, substream, msg);
return Poll::Pending;
}
Poll::Ready(Err(_)) => return Poll::Ready(None),
}
}
InboundSubstreamState::PendingFlush(id, mut substream) => {
match substream.poll_flush_unpin(cx) {
Poll::Ready(Ok(())) => {
*this = InboundSubstreamState::WaitingMessage {
first: false,
connection_id: id,
substream,
};
}
Poll::Pending => {
*this = InboundSubstreamState::PendingFlush(id, substream);
return Poll::Pending;
}
Poll::Ready(Err(_)) => return Poll::Ready(None),
}
}
InboundSubstreamState::Closing(mut stream) => match stream.poll_close_unpin(cx) {
Poll::Ready(Ok(())) | Poll::Ready(Err(_)) => return Poll::Ready(None),
Poll::Pending => {
*this = InboundSubstreamState::Closing(stream);
return Poll::Pending;
}
},
InboundSubstreamState::Poisoned { .. } => unreachable!(),
InboundSubstreamState::Cancelled => return Poll::Ready(None),
}
}
}
}
fn process_kad_response<TUserData>(
event: KadResponseMsg,
user_data: TUserData,
) -> KademliaHandlerEvent<TUserData> {
match event {
KadResponseMsg::Pong => {
KademliaHandlerEvent::QueryError {
error: KademliaHandlerQueryErr::UnexpectedMessage,
user_data,
}
}
KadResponseMsg::FindNode { closer_peers } => KademliaHandlerEvent::FindNodeRes {
closer_peers,
user_data,
},
KadResponseMsg::GetProviders {
closer_peers,
provider_peers,
} => KademliaHandlerEvent::GetProvidersRes {
closer_peers,
provider_peers,
user_data,
},
KadResponseMsg::GetValue {
record,
closer_peers,
} => KademliaHandlerEvent::GetRecordRes {
record,
closer_peers,
user_data,
},
KadResponseMsg::PutValue { key, value, .. } => KademliaHandlerEvent::PutRecordRes {
key,
value,
user_data,
},
}
}