use crate::{
discovery::{DiscoveryBehaviour, DiscoveryConfig, DiscoveryOut},
event::DhtEvent,
peer_info,
peer_store::PeerStoreProvider,
protocol::{CustomMessageOutcome, NotificationsSink, Protocol},
protocol_controller::SetId,
request_responses::{self, IfDisconnected, ProtocolConfig, RequestFailure},
service::traits::Direction,
types::ProtocolName,
ReputationChange,
};
use futures::channel::oneshot;
use libp2p::{
connection_limits::ConnectionLimits,
core::Multiaddr,
identify::Info as IdentifyInfo,
identity::PublicKey,
kad::{Record, RecordKey},
swarm::NetworkBehaviour,
PeerId, StreamProtocol,
};
use parking_lot::Mutex;
use sp_runtime::traits::Block as BlockT;
use std::{
collections::HashSet,
sync::Arc,
time::{Duration, Instant},
};
pub use crate::request_responses::{InboundFailure, OutboundFailure, ResponseFailure};
#[derive(NetworkBehaviour)]
#[behaviour(to_swarm = "BehaviourOut")]
pub struct Behaviour<B: BlockT> {
connection_limits: libp2p::connection_limits::Behaviour,
substrate: Protocol<B>,
peer_info: peer_info::PeerInfoBehaviour,
discovery: DiscoveryBehaviour,
request_responses: request_responses::RequestResponsesBehaviour,
}
pub enum BehaviourOut {
RandomKademliaStarted,
InboundRequest {
protocol: ProtocolName,
result: Result<Duration, ResponseFailure>,
},
RequestFinished {
protocol: ProtocolName,
duration: Duration,
result: Result<(), RequestFailure>,
},
ReputationChanges { peer: PeerId, changes: Vec<ReputationChange> },
NotificationStreamOpened {
remote: PeerId,
set_id: SetId,
direction: Direction,
negotiated_fallback: Option<ProtocolName>,
notifications_sink: NotificationsSink,
received_handshake: Vec<u8>,
},
NotificationStreamReplaced {
remote: PeerId,
set_id: SetId,
notifications_sink: NotificationsSink,
},
NotificationStreamClosed {
remote: PeerId,
set_id: SetId,
},
NotificationsReceived {
remote: PeerId,
set_id: SetId,
notification: Vec<u8>,
},
PeerIdentify {
peer_id: PeerId,
info: IdentifyInfo,
},
Discovered(PeerId),
Dht(DhtEvent, Option<Duration>),
None,
}
impl<B: BlockT> Behaviour<B> {
pub fn new(
substrate: Protocol<B>,
user_agent: String,
local_public_key: PublicKey,
disco_config: DiscoveryConfig,
request_response_protocols: Vec<ProtocolConfig>,
peer_store_handle: Arc<dyn PeerStoreProvider>,
external_addresses: Arc<Mutex<HashSet<Multiaddr>>>,
connection_limits: ConnectionLimits,
) -> Result<Self, request_responses::RegisterError> {
Ok(Self {
substrate,
peer_info: peer_info::PeerInfoBehaviour::new(
user_agent,
local_public_key,
external_addresses,
),
discovery: disco_config.finish(),
request_responses: request_responses::RequestResponsesBehaviour::new(
request_response_protocols.into_iter(),
peer_store_handle,
)?,
connection_limits: libp2p::connection_limits::Behaviour::new(connection_limits),
})
}
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) -> Option<Vec<(u32, usize)>> {
self.discovery.num_entries_per_kbucket()
}
pub fn num_kademlia_records(&mut self) -> Option<usize> {
self.discovery.num_kademlia_records()
}
pub fn kademlia_records_total_size(&mut self) -> Option<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: ProtocolName,
request: Vec<u8>,
fallback_request: Option<(Vec<u8>, ProtocolName)>,
pending_response: oneshot::Sender<Result<(Vec<u8>, ProtocolName), RequestFailure>>,
connect: IfDisconnected,
) {
self.request_responses.send_request(
target,
protocol,
request,
fallback_request,
pending_response,
connect,
)
}
pub fn user_protocol(&self) -> &Protocol<B> {
&self.substrate
}
pub fn user_protocol_mut(&mut self) -> &mut Protocol<B> {
&mut self.substrate
}
pub fn add_self_reported_address_to_dht(
&mut self,
peer_id: &PeerId,
supported_protocols: &[StreamProtocol],
addr: Multiaddr,
) {
self.discovery.add_self_reported_address(peer_id, supported_protocols, addr);
}
pub fn get_value(&mut self, key: RecordKey) {
self.discovery.get_value(key);
}
pub fn put_value(&mut self, key: RecordKey, value: Vec<u8>) {
self.discovery.put_value(key, value);
}
pub fn put_record_to(
&mut self,
record: Record,
peers: HashSet<sc_network_types::PeerId>,
update_local_storage: bool,
) {
self.discovery.put_record_to(record, peers, update_local_storage);
}
pub fn store_record(
&mut self,
record_key: RecordKey,
record_value: Vec<u8>,
publisher: Option<PeerId>,
expires: Option<Instant>,
) {
self.discovery.store_record(record_key, record_value, publisher, expires);
}
}
impl From<CustomMessageOutcome> for BehaviourOut {
fn from(event: CustomMessageOutcome) -> Self {
match event {
CustomMessageOutcome::NotificationStreamOpened {
remote,
set_id,
direction,
negotiated_fallback,
received_handshake,
notifications_sink,
} => BehaviourOut::NotificationStreamOpened {
remote,
set_id,
direction,
negotiated_fallback,
received_handshake,
notifications_sink,
},
CustomMessageOutcome::NotificationStreamReplaced {
remote,
set_id,
notifications_sink,
} => BehaviourOut::NotificationStreamReplaced { remote, set_id, notifications_sink },
CustomMessageOutcome::NotificationStreamClosed { remote, set_id } =>
BehaviourOut::NotificationStreamClosed { remote, set_id },
CustomMessageOutcome::NotificationsReceived { remote, set_id, notification } =>
BehaviourOut::NotificationsReceived { remote, set_id, notification },
}
}
}
impl From<request_responses::Event> for BehaviourOut {
fn from(event: request_responses::Event) -> Self {
match event {
request_responses::Event::InboundRequest { protocol, result, .. } =>
BehaviourOut::InboundRequest { protocol, result },
request_responses::Event::RequestFinished { protocol, duration, result, .. } =>
BehaviourOut::RequestFinished { protocol, duration, result },
request_responses::Event::ReputationChanges { peer, changes } =>
BehaviourOut::ReputationChanges { peer, changes },
}
}
}
impl From<peer_info::PeerInfoEvent> for BehaviourOut {
fn from(event: peer_info::PeerInfoEvent) -> Self {
let peer_info::PeerInfoEvent::Identified { peer_id, info } = event;
BehaviourOut::PeerIdentify { peer_id, info }
}
}
impl From<DiscoveryOut> for BehaviourOut {
fn from(event: DiscoveryOut) -> Self {
match event {
DiscoveryOut::UnroutablePeer(_peer_id) => {
BehaviourOut::None
},
DiscoveryOut::Discovered(peer_id) => BehaviourOut::Discovered(peer_id),
DiscoveryOut::ValueFound(results, duration) =>
BehaviourOut::Dht(DhtEvent::ValueFound(results), Some(duration)),
DiscoveryOut::ValueNotFound(key, duration) =>
BehaviourOut::Dht(DhtEvent::ValueNotFound(key), Some(duration)),
DiscoveryOut::ValuePut(key, duration) =>
BehaviourOut::Dht(DhtEvent::ValuePut(key), Some(duration)),
DiscoveryOut::PutRecordRequest(record_key, record_value, publisher, expires) =>
BehaviourOut::Dht(
DhtEvent::PutRecordRequest(record_key, record_value, publisher, expires),
None,
),
DiscoveryOut::ValuePutFailed(key, duration) =>
BehaviourOut::Dht(DhtEvent::ValuePutFailed(key), Some(duration)),
DiscoveryOut::RandomKademliaStarted => BehaviourOut::RandomKademliaStarted,
}
}
}
impl From<void::Void> for BehaviourOut {
fn from(e: void::Void) -> Self {
void::unreachable(e)
}
}