use crate::{
config::{ProtocolId, Role},
bitswap::Bitswap,
discovery::{DiscoveryBehaviour, DiscoveryConfig, DiscoveryOut},
protocol::{message::Roles, CustomMessageOutcome, NotificationsSink, Protocol},
peer_info, request_responses, light_client_requests,
ObservedRole, DhtEvent, ExHashT,
};
use bytes::Bytes;
use futures::{channel::oneshot, stream::StreamExt};
use libp2p::NetworkBehaviour;
use libp2p::core::{Multiaddr, PeerId, PublicKey};
use libp2p::identify::IdentifyInfo;
use libp2p::kad::record;
use libp2p::swarm::{
NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters, toggle::Toggle
};
use log::debug;
use prost::Message;
use sp_consensus::{BlockOrigin, import_queue::{IncomingBlock, Origin}};
use sp_runtime::{traits::{Block as BlockT, NumberFor}, Justification};
use std::{
borrow::Cow,
collections::{HashSet, VecDeque},
iter,
task::{Context, Poll},
time::Duration,
};
pub use crate::request_responses::{
ResponseFailure, InboundFailure, RequestFailure, OutboundFailure, RequestId,
IfDisconnected
};
#[derive(NetworkBehaviour)]
#[behaviour(out_event = "BehaviourOut<B>", poll_method = "poll")]
pub struct Behaviour<B: BlockT, H: ExHashT> {
substrate: Protocol<B, H>,
peer_info: peer_info::PeerInfoBehaviour,
discovery: DiscoveryBehaviour,
bitswap: Toggle<Bitswap<B>>,
request_responses: request_responses::RequestResponsesBehaviour,
#[behaviour(ignore)]
events: VecDeque<BehaviourOut<B>>,
#[behaviour(ignore)]
role: Role,
#[behaviour(ignore)]
light_client_request_sender: light_client_requests::sender::LightClientRequestSender<B>,
#[behaviour(ignore)]
block_request_protocol_name: String,
}
pub enum BehaviourOut<B: BlockT> {
BlockImport(BlockOrigin, Vec<IncomingBlock<B>>),
JustificationImport(Origin, B::Hash, NumberFor<B>, Justification),
RandomKademliaStarted(ProtocolId),
InboundRequest {
peer: PeerId,
protocol: Cow<'static, str>,
result: Result<Duration, ResponseFailure>,
},
RequestFinished {
peer: PeerId,
protocol: Cow<'static, str>,
duration: Duration,
result: Result<(), RequestFailure>,
},
NotificationStreamOpened {
remote: PeerId,
protocol: Cow<'static, str>,
notifications_sink: NotificationsSink,
role: ObservedRole,
},
NotificationStreamReplaced {
remote: PeerId,
protocol: Cow<'static, str>,
notifications_sink: NotificationsSink,
},
NotificationStreamClosed {
remote: PeerId,
protocol: Cow<'static, str>,
},
NotificationsReceived {
remote: PeerId,
messages: Vec<(Cow<'static, str>, Bytes)>,
},
SyncConnected(PeerId),
SyncDisconnected(PeerId),
Dht(DhtEvent, Duration),
}
impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
pub fn new(
substrate: Protocol<B, H>,
role: Role,
user_agent: String,
local_public_key: PublicKey,
light_client_request_sender: light_client_requests::sender::LightClientRequestSender<B>,
disco_config: DiscoveryConfig,
block_request_protocol_config: request_responses::ProtocolConfig,
bitswap: Option<Bitswap<B>>,
light_client_request_protocol_config: request_responses::ProtocolConfig,
mut request_response_protocols: Vec<request_responses::ProtocolConfig>,
) -> Result<Self, request_responses::RegisterError> {
let block_request_protocol_name = block_request_protocol_config.name.to_string();
request_response_protocols.push(block_request_protocol_config);
request_response_protocols.push(light_client_request_protocol_config);
Ok(Behaviour {
substrate,
peer_info: peer_info::PeerInfoBehaviour::new(user_agent, local_public_key),
discovery: disco_config.finish(),
bitswap: bitswap.into(),
request_responses:
request_responses::RequestResponsesBehaviour::new(request_response_protocols.into_iter())?,
light_client_request_sender,
events: VecDeque::new(),
role,
block_request_protocol_name,
})
}
pub fn known_peers(&mut self) -> HashSet<PeerId> {
self.discovery.known_peers()
}
pub fn add_known_address(&mut self, peer_id: PeerId, addr: Multiaddr) {
self.discovery.add_known_address(peer_id, addr)
}
pub fn num_entries_per_kbucket(&mut self) -> impl ExactSizeIterator<Item = (&ProtocolId, Vec<(u32, usize)>)> {
self.discovery.num_entries_per_kbucket()
}
pub fn num_kademlia_records(&mut self) -> impl ExactSizeIterator<Item = (&ProtocolId, usize)> {
self.discovery.num_kademlia_records()
}
pub fn kademlia_records_total_size(&mut self) -> impl ExactSizeIterator<Item = (&ProtocolId, usize)> {
self.discovery.kademlia_records_total_size()
}
pub fn node(&self, peer_id: &PeerId) -> Option<peer_info::Node> {
self.peer_info.node(peer_id)
}
pub fn send_request(
&mut self,
target: &PeerId,
protocol: &str,
request: Vec<u8>,
pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
connect: IfDisconnected,
) {
self.request_responses.send_request(target, protocol, request, pending_response, connect)
}
pub fn user_protocol(&self) -> &Protocol<B, H> {
&self.substrate
}
pub fn user_protocol_mut(&mut self) -> &mut Protocol<B, H> {
&mut self.substrate
}
pub fn get_value(&mut self, key: &record::Key) {
self.discovery.get_value(key);
}
pub fn put_value(&mut self, key: record::Key, value: Vec<u8>) {
self.discovery.put_value(key, value);
}
pub fn light_client_request(
&mut self,
r: light_client_requests::sender::Request<B>,
) -> Result<(), light_client_requests::sender::SendRequestError> {
self.light_client_request_sender.request(r)
}
}
fn reported_roles_to_observed_role(local_role: &Role, remote: &PeerId, roles: Roles) -> ObservedRole {
if roles.is_authority() {
match local_role {
Role::Authority { sentry_nodes }
if sentry_nodes.iter().any(|s| s.peer_id == *remote) => ObservedRole::OurSentry,
Role::Sentry { validators }
if validators.iter().any(|s| s.peer_id == *remote) => ObservedRole::OurGuardedAuthority,
_ => ObservedRole::Authority
}
} else if roles.is_full() {
ObservedRole::Full
} else {
ObservedRole::Light
}
}
impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<void::Void> for
Behaviour<B, H> {
fn inject_event(&mut self, event: void::Void) {
void::unreachable(event)
}
}
impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<CustomMessageOutcome<B>> for
Behaviour<B, H> {
fn inject_event(&mut self, event: CustomMessageOutcome<B>) {
match event {
CustomMessageOutcome::BlockImport(origin, blocks) =>
self.events.push_back(BehaviourOut::BlockImport(origin, blocks)),
CustomMessageOutcome::JustificationImport(origin, hash, nb, justification) =>
self.events.push_back(BehaviourOut::JustificationImport(origin, hash, nb, justification)),
CustomMessageOutcome::BlockRequest { target, request, pending_response } => {
let mut buf = Vec::with_capacity(request.encoded_len());
if let Err(err) = request.encode(&mut buf) {
log::warn!(
target: "sync",
"Failed to encode block request {:?}: {:?}",
request, err
);
return
}
self.request_responses.send_request(
&target, &self.block_request_protocol_name, buf, pending_response, IfDisconnected::ImmediateError,
);
},
CustomMessageOutcome::NotificationStreamOpened { remote, protocol, roles, notifications_sink } => {
let role = reported_roles_to_observed_role(&self.role, &remote, roles);
self.events.push_back(BehaviourOut::NotificationStreamOpened {
remote,
protocol,
role: role.clone(),
notifications_sink: notifications_sink.clone(),
});
},
CustomMessageOutcome::NotificationStreamReplaced { remote, protocol, notifications_sink } =>
self.events.push_back(BehaviourOut::NotificationStreamReplaced {
remote,
protocol,
notifications_sink,
}),
CustomMessageOutcome::NotificationStreamClosed { remote, protocol } =>
self.events.push_back(BehaviourOut::NotificationStreamClosed {
remote,
protocol,
}),
CustomMessageOutcome::NotificationsReceived { remote, messages } => {
self.events.push_back(BehaviourOut::NotificationsReceived { remote, messages });
},
CustomMessageOutcome::PeerNewBest(peer_id, number) => {
self.light_client_request_sender.update_best_block(&peer_id, number);
}
CustomMessageOutcome::SyncConnected(peer_id) => {
self.light_client_request_sender.inject_connected(peer_id);
self.events.push_back(BehaviourOut::SyncConnected(peer_id))
}
CustomMessageOutcome::SyncDisconnected(peer_id) => {
self.light_client_request_sender.inject_disconnected(peer_id);
self.events.push_back(BehaviourOut::SyncDisconnected(peer_id))
}
CustomMessageOutcome::None => {}
}
}
}
impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<request_responses::Event> for Behaviour<B, H> {
fn inject_event(&mut self, event: request_responses::Event) {
match event {
request_responses::Event::InboundRequest { peer, protocol, result } => {
self.events.push_back(BehaviourOut::InboundRequest {
peer,
protocol,
result,
});
}
request_responses::Event::RequestFinished { peer, protocol, duration, result } => {
self.events.push_back(BehaviourOut::RequestFinished {
peer, protocol, duration, result,
});
},
request_responses::Event::ReputationChanges { peer, changes } => {
for change in changes {
self.substrate.report_peer(peer, change);
}
}
}
}
}
impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<peer_info::PeerInfoEvent>
for Behaviour<B, H> {
fn inject_event(&mut self, event: peer_info::PeerInfoEvent) {
let peer_info::PeerInfoEvent::Identified {
peer_id,
info: IdentifyInfo {
protocol_version,
agent_version,
mut listen_addrs,
protocols,
..
},
} = event;
if listen_addrs.len() > 30 {
debug!(
target: "sub-libp2p",
"Node {:?} has reported more than 30 addresses; it is identified by {:?} and {:?}",
peer_id, protocol_version, agent_version
);
listen_addrs.truncate(30);
}
for addr in listen_addrs {
self.discovery.add_self_reported_address(&peer_id, protocols.iter(), addr);
}
self.substrate.add_default_set_discovered_nodes(iter::once(peer_id));
}
}
impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<DiscoveryOut>
for Behaviour<B, H> {
fn inject_event(&mut self, out: DiscoveryOut) {
match out {
DiscoveryOut::UnroutablePeer(_peer_id) => {
}
DiscoveryOut::Discovered(peer_id) => {
self.substrate.add_default_set_discovered_nodes(iter::once(peer_id));
}
DiscoveryOut::ValueFound(results, duration) => {
self.events.push_back(BehaviourOut::Dht(DhtEvent::ValueFound(results), duration));
}
DiscoveryOut::ValueNotFound(key, duration) => {
self.events.push_back(BehaviourOut::Dht(DhtEvent::ValueNotFound(key), duration));
}
DiscoveryOut::ValuePut(key, duration) => {
self.events.push_back(BehaviourOut::Dht(DhtEvent::ValuePut(key), duration));
}
DiscoveryOut::ValuePutFailed(key, duration) => {
self.events.push_back(BehaviourOut::Dht(DhtEvent::ValuePutFailed(key), duration));
}
DiscoveryOut::RandomKademliaStarted(protocols) => {
for protocol in protocols {
self.events.push_back(BehaviourOut::RandomKademliaStarted(protocol));
}
}
}
}
}
impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
fn poll<TEv>(
&mut self,
cx: &mut Context,
_: &mut impl PollParameters,
) -> Poll<NetworkBehaviourAction<TEv, BehaviourOut<B>>> {
use light_client_requests::sender::OutEvent;
while let Poll::Ready(Some(event)) =
self.light_client_request_sender.poll_next_unpin(cx)
{
match event {
OutEvent::SendRequest {
target,
request,
pending_response,
protocol_name,
} => self.request_responses.send_request(
&target,
&protocol_name,
request,
pending_response,
IfDisconnected::ImmediateError,
),
}
}
if let Some(event) = self.events.pop_front() {
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event))
}
Poll::Pending
}
}