mirror of
https://github.com/pezkuwichain/pezkuwi-subxt.git
synced 2026-05-09 12:58:05 +00:00
Use custom type for ProtocolName (#12172)
* Add ProtocolName custom type * Use new ProtocolName in sc_network_common * Use new ProtocolName in sc_network * Use new ProtocolName for BEEFY and GRANDPA * Use new ProtocolName for notifications * Use new ProtocolName in sc_network (part 2) * Use new ProtocolName in sc_network_gossip * Use new ProtocolName in sc_offchain * Remove unused imports * Some more fixes * Add tests * Fix minor import issues * Re-export ProtocolName in sc_network * Revert "Re-export ProtocolName in sc_network" This reverts commit 8d8ff71927e7750757f29c9bbd88dc0ba181d214. * Re-export ProtocolName in sc_network * Remove dependency on sc-network-common from beefy-gadget
This commit is contained in:
@@ -42,7 +42,10 @@ use log::debug;
|
||||
use sc_consensus::import_queue::{IncomingBlock, Origin};
|
||||
use sc_network_common::{
|
||||
config::ProtocolId,
|
||||
protocol::event::{DhtEvent, ObservedRole},
|
||||
protocol::{
|
||||
event::{DhtEvent, ObservedRole},
|
||||
ProtocolName,
|
||||
},
|
||||
request_responses::{IfDisconnected, ProtocolConfig, RequestFailure},
|
||||
};
|
||||
use sc_peerset::PeersetHandle;
|
||||
@@ -53,7 +56,6 @@ use sp_runtime::{
|
||||
Justifications,
|
||||
};
|
||||
use std::{
|
||||
borrow::Cow,
|
||||
collections::{HashSet, VecDeque},
|
||||
iter,
|
||||
task::{Context, Poll},
|
||||
@@ -117,7 +119,7 @@ pub enum BehaviourOut<B: BlockT> {
|
||||
/// Peer which sent us a request.
|
||||
peer: PeerId,
|
||||
/// Protocol name of the request.
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
/// If `Ok`, contains the time elapsed between when we received the request and when we
|
||||
/// sent back the response. If `Err`, the error that happened.
|
||||
result: Result<Duration, ResponseFailure>,
|
||||
@@ -130,7 +132,7 @@ pub enum BehaviourOut<B: BlockT> {
|
||||
/// Peer that we send a request to.
|
||||
peer: PeerId,
|
||||
/// Name of the protocol in question.
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
/// Duration the request took.
|
||||
duration: Duration,
|
||||
/// Result of the request.
|
||||
@@ -144,12 +146,12 @@ pub enum BehaviourOut<B: BlockT> {
|
||||
/// Node we opened the substream with.
|
||||
remote: PeerId,
|
||||
/// The concerned protocol. Each protocol uses a different substream.
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
/// If the negotiation didn't use the main name of the protocol (the one in
|
||||
/// `notifications_protocol`), then this field contains which name has actually been
|
||||
/// used.
|
||||
/// See also [`crate::Event::NotificationStreamOpened`].
|
||||
negotiated_fallback: Option<Cow<'static, str>>,
|
||||
negotiated_fallback: Option<ProtocolName>,
|
||||
/// Object that permits sending notifications to the peer.
|
||||
notifications_sink: NotificationsSink,
|
||||
/// Role of the remote.
|
||||
@@ -165,7 +167,7 @@ pub enum BehaviourOut<B: BlockT> {
|
||||
/// Id of the peer we are connected to.
|
||||
remote: PeerId,
|
||||
/// The concerned protocol. Each protocol uses a different substream.
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
/// Replacement for the previous [`NotificationsSink`].
|
||||
notifications_sink: NotificationsSink,
|
||||
},
|
||||
@@ -176,7 +178,7 @@ pub enum BehaviourOut<B: BlockT> {
|
||||
/// Node we closed the substream with.
|
||||
remote: PeerId,
|
||||
/// The concerned protocol. Each protocol uses a different substream.
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
},
|
||||
|
||||
/// Received one or more messages from the given node using the given protocol.
|
||||
@@ -184,7 +186,7 @@ pub enum BehaviourOut<B: BlockT> {
|
||||
/// Node we received the message from.
|
||||
remote: PeerId,
|
||||
/// Concerned protocol and associated message.
|
||||
messages: Vec<(Cow<'static, str>, Bytes)>,
|
||||
messages: Vec<(ProtocolName, Bytes)>,
|
||||
},
|
||||
|
||||
/// Now connected to a new peer for syncing purposes.
|
||||
|
||||
@@ -41,10 +41,9 @@ use libp2p::{
|
||||
};
|
||||
use prometheus_endpoint::Registry;
|
||||
use sc_consensus::ImportQueue;
|
||||
use sc_network_common::{config::MultiaddrWithPeerId, sync::ChainSync};
|
||||
use sc_network_common::{config::MultiaddrWithPeerId, protocol::ProtocolName, sync::ChainSync};
|
||||
use sp_runtime::traits::Block as BlockT;
|
||||
use std::{
|
||||
borrow::Cow,
|
||||
collections::HashMap,
|
||||
error::Error,
|
||||
fs,
|
||||
@@ -431,14 +430,14 @@ pub struct NonDefaultSetConfig {
|
||||
///
|
||||
/// > **Note**: This field isn't present for the default set, as this is handled internally
|
||||
/// > by the networking code.
|
||||
pub notifications_protocol: Cow<'static, str>,
|
||||
pub notifications_protocol: ProtocolName,
|
||||
/// If the remote reports that it doesn't support the protocol indicated in the
|
||||
/// `notifications_protocol` field, then each of these fallback names will be tried one by
|
||||
/// one.
|
||||
///
|
||||
/// If a fallback is used, it will be reported in
|
||||
/// [`crate::Event::NotificationStreamOpened::negotiated_fallback`].
|
||||
pub fallback_names: Vec<Cow<'static, str>>,
|
||||
pub fallback_names: Vec<ProtocolName>,
|
||||
/// Maximum allowed size of single notifications.
|
||||
pub max_notification_size: u64,
|
||||
/// Base configuration.
|
||||
@@ -447,7 +446,7 @@ pub struct NonDefaultSetConfig {
|
||||
|
||||
impl NonDefaultSetConfig {
|
||||
/// Creates a new [`NonDefaultSetConfig`]. Zero slots and accepts only reserved nodes.
|
||||
pub fn new(notifications_protocol: Cow<'static, str>, max_notification_size: u64) -> Self {
|
||||
pub fn new(notifications_protocol: ProtocolName, max_notification_size: u64) -> Self {
|
||||
Self {
|
||||
notifications_protocol,
|
||||
max_notification_size,
|
||||
@@ -476,7 +475,7 @@ impl NonDefaultSetConfig {
|
||||
/// Add a list of protocol names used for backward compatibility.
|
||||
///
|
||||
/// See the explanations in [`NonDefaultSetConfig::fallback_names`].
|
||||
pub fn add_fallback_names(&mut self, fallback_names: Vec<Cow<'static, str>>) {
|
||||
pub fn add_fallback_names(&mut self, fallback_names: Vec<ProtocolName>) {
|
||||
self.fallback_names.extend(fallback_names);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,8 +20,9 @@
|
||||
|
||||
use crate::config::TransportConfig;
|
||||
use libp2p::{Multiaddr, PeerId};
|
||||
use sc_network_common::protocol::ProtocolName;
|
||||
|
||||
use std::{borrow::Cow, fmt};
|
||||
use std::fmt;
|
||||
|
||||
/// Result type alias for the network.
|
||||
pub type Result<T> = std::result::Result<T, Error>;
|
||||
@@ -65,7 +66,7 @@ pub enum Error {
|
||||
#[error("Request-response protocol registered multiple times: {protocol}")]
|
||||
DuplicateRequestResponseProtocol {
|
||||
/// Name of the protocol registered multiple times.
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@@ -264,7 +264,10 @@ pub mod transactions;
|
||||
pub use libp2p::{multiaddr, Multiaddr, PeerId};
|
||||
pub use protocol::PeerInfo;
|
||||
pub use sc_network_common::{
|
||||
protocol::event::{DhtEvent, Event, ObservedRole},
|
||||
protocol::{
|
||||
event::{DhtEvent, Event, ObservedRole},
|
||||
ProtocolName,
|
||||
},
|
||||
request_responses::{IfDisconnected, RequestFailure},
|
||||
service::{
|
||||
KademliaKey, NetworkBlock, NetworkDHTProvider, NetworkRequest, NetworkSigner,
|
||||
|
||||
@@ -44,6 +44,7 @@ use sc_client_api::HeaderBackend;
|
||||
use sc_consensus::import_queue::{BlockImportError, BlockImportStatus, IncomingBlock, Origin};
|
||||
use sc_network_common::{
|
||||
config::ProtocolId,
|
||||
protocol::ProtocolName,
|
||||
request_responses::RequestFailure,
|
||||
sync::{
|
||||
message::{
|
||||
@@ -63,7 +64,6 @@ use sp_runtime::{
|
||||
Justifications,
|
||||
};
|
||||
use std::{
|
||||
borrow::Cow,
|
||||
collections::{HashMap, HashSet, VecDeque},
|
||||
io, iter,
|
||||
num::NonZeroUsize,
|
||||
@@ -190,7 +190,7 @@ pub struct Protocol<B: BlockT, Client> {
|
||||
/// Handles opening the unique substream and sending and receiving raw messages.
|
||||
behaviour: Notifications,
|
||||
/// List of notifications protocols that have been registered.
|
||||
notification_protocols: Vec<Cow<'static, str>>,
|
||||
notification_protocols: Vec<ProtocolName>,
|
||||
/// If we receive a new "substream open" event that contains an invalid handshake, we ask the
|
||||
/// inner layer to force-close the substream. Force-closing the substream will generate a
|
||||
/// "substream closed" event. This is a problem: since we can't propagate the "substream open"
|
||||
@@ -460,7 +460,7 @@ where
|
||||
}
|
||||
|
||||
/// Disconnects the given peer if we are connected to it.
|
||||
pub fn disconnect_peer(&mut self, peer_id: &PeerId, protocol_name: &str) {
|
||||
pub fn disconnect_peer(&mut self, peer_id: &PeerId, protocol_name: ProtocolName) {
|
||||
if let Some(position) = self.notification_protocols.iter().position(|p| *p == protocol_name)
|
||||
{
|
||||
self.behaviour.disconnect_peer(
|
||||
@@ -1085,7 +1085,7 @@ where
|
||||
}
|
||||
|
||||
/// Sets the list of reserved peers for the given protocol/peerset.
|
||||
pub fn set_reserved_peerset_peers(&self, protocol: Cow<'static, str>, peers: HashSet<PeerId>) {
|
||||
pub fn set_reserved_peerset_peers(&self, protocol: ProtocolName, peers: HashSet<PeerId>) {
|
||||
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
|
||||
self.peerset_handle
|
||||
.set_reserved_peers(sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS), peers);
|
||||
@@ -1099,7 +1099,7 @@ where
|
||||
}
|
||||
|
||||
/// Removes a `PeerId` from the list of reserved peers.
|
||||
pub fn remove_set_reserved_peer(&self, protocol: Cow<'static, str>, peer: PeerId) {
|
||||
pub fn remove_set_reserved_peer(&self, protocol: ProtocolName, peer: PeerId) {
|
||||
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
|
||||
self.peerset_handle.remove_reserved_peer(
|
||||
sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS),
|
||||
@@ -1115,7 +1115,7 @@ where
|
||||
}
|
||||
|
||||
/// Adds a `PeerId` to the list of reserved peers.
|
||||
pub fn add_set_reserved_peer(&self, protocol: Cow<'static, str>, peer: PeerId) {
|
||||
pub fn add_set_reserved_peer(&self, protocol: ProtocolName, peer: PeerId) {
|
||||
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
|
||||
self.peerset_handle
|
||||
.add_reserved_peer(sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS), peer);
|
||||
@@ -1138,7 +1138,7 @@ where
|
||||
}
|
||||
|
||||
/// Add a peer to a peers set.
|
||||
pub fn add_to_peers_set(&self, protocol: Cow<'static, str>, peer: PeerId) {
|
||||
pub fn add_to_peers_set(&self, protocol: ProtocolName, peer: PeerId) {
|
||||
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
|
||||
self.peerset_handle
|
||||
.add_to_peers_set(sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS), peer);
|
||||
@@ -1152,7 +1152,7 @@ where
|
||||
}
|
||||
|
||||
/// Remove a peer from a peers set.
|
||||
pub fn remove_from_peers_set(&self, protocol: Cow<'static, str>, peer: PeerId) {
|
||||
pub fn remove_from_peers_set(&self, protocol: ProtocolName, peer: PeerId) {
|
||||
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
|
||||
self.peerset_handle.remove_from_peers_set(
|
||||
sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS),
|
||||
@@ -1259,27 +1259,27 @@ pub enum CustomMessageOutcome<B: BlockT> {
|
||||
/// Notification protocols have been opened with a remote.
|
||||
NotificationStreamOpened {
|
||||
remote: PeerId,
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
/// See [`crate::Event::NotificationStreamOpened::negotiated_fallback`].
|
||||
negotiated_fallback: Option<Cow<'static, str>>,
|
||||
negotiated_fallback: Option<ProtocolName>,
|
||||
roles: Roles,
|
||||
notifications_sink: NotificationsSink,
|
||||
},
|
||||
/// The [`NotificationsSink`] of some notification protocols need an update.
|
||||
NotificationStreamReplaced {
|
||||
remote: PeerId,
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
notifications_sink: NotificationsSink,
|
||||
},
|
||||
/// Notification protocols have been closed with a remote.
|
||||
NotificationStreamClosed {
|
||||
remote: PeerId,
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
},
|
||||
/// Messages have been received on one or more notifications protocols.
|
||||
NotificationsReceived {
|
||||
remote: PeerId,
|
||||
messages: Vec<(Cow<'static, str>, Bytes)>,
|
||||
messages: Vec<(ProtocolName, Bytes)>,
|
||||
},
|
||||
/// A new block request must be emitted.
|
||||
BlockRequest {
|
||||
|
||||
@@ -33,15 +33,14 @@ use libp2p::{
|
||||
use log::{error, trace, warn};
|
||||
use parking_lot::RwLock;
|
||||
use rand::distributions::{Distribution as _, Uniform};
|
||||
use sc_network_common::protocol::ProtocolName;
|
||||
use sc_peerset::DropReason;
|
||||
use smallvec::SmallVec;
|
||||
use std::{
|
||||
borrow::Cow,
|
||||
cmp,
|
||||
collections::{hash_map::Entry, VecDeque},
|
||||
mem,
|
||||
pin::Pin,
|
||||
str,
|
||||
sync::Arc,
|
||||
task::{Context, Poll},
|
||||
time::{Duration, Instant},
|
||||
@@ -140,9 +139,9 @@ pub struct Notifications {
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ProtocolConfig {
|
||||
/// Name of the protocol.
|
||||
pub name: Cow<'static, str>,
|
||||
pub name: ProtocolName,
|
||||
/// Names of the protocol to use if the main one isn't available.
|
||||
pub fallback_names: Vec<Cow<'static, str>>,
|
||||
pub fallback_names: Vec<ProtocolName>,
|
||||
/// Handshake of the protocol.
|
||||
pub handshake: Vec<u8>,
|
||||
/// Maximum allowed size for a notification.
|
||||
@@ -309,7 +308,7 @@ pub enum NotificationsOut {
|
||||
set_id: sc_peerset::SetId,
|
||||
/// If `Some`, a fallback protocol name has been used rather the main protocol name.
|
||||
/// Always matches one of the fallback names passed at initialization.
|
||||
negotiated_fallback: Option<Cow<'static, str>>,
|
||||
negotiated_fallback: Option<ProtocolName>,
|
||||
/// Handshake that was sent to us.
|
||||
/// This is normally a "Status" message, but this is out of the concern of this code.
|
||||
received_handshake: Vec<u8>,
|
||||
|
||||
@@ -80,12 +80,11 @@ use libp2p::{
|
||||
};
|
||||
use log::error;
|
||||
use parking_lot::{Mutex, RwLock};
|
||||
use sc_network_common::protocol::ProtocolName;
|
||||
use std::{
|
||||
borrow::Cow,
|
||||
collections::VecDeque,
|
||||
mem,
|
||||
pin::Pin,
|
||||
str,
|
||||
sync::Arc,
|
||||
task::{Context, Poll},
|
||||
time::{Duration, Instant},
|
||||
@@ -146,9 +145,9 @@ pub struct NotifsHandler {
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ProtocolConfig {
|
||||
/// Name of the protocol.
|
||||
pub name: Cow<'static, str>,
|
||||
pub name: ProtocolName,
|
||||
/// Names of the protocol to use if the main one isn't available.
|
||||
pub fallback_names: Vec<Cow<'static, str>>,
|
||||
pub fallback_names: Vec<ProtocolName>,
|
||||
/// Handshake of the protocol. The `RwLock` is locked every time a new substream is opened.
|
||||
pub handshake: Arc<RwLock<Vec<u8>>>,
|
||||
/// Maximum allowed size for a notification.
|
||||
@@ -297,7 +296,7 @@ pub enum NotifsHandlerOut {
|
||||
/// Index of the protocol in the list of protocols passed at initialization.
|
||||
protocol_index: usize,
|
||||
/// Name of the protocol that was actually negotiated, if the default one wasn't available.
|
||||
negotiated_fallback: Option<Cow<'static, str>>,
|
||||
negotiated_fallback: Option<ProtocolName>,
|
||||
/// The endpoint of the connection that is open for custom protocols.
|
||||
endpoint: ConnectedPoint,
|
||||
/// Handshake that was sent to us.
|
||||
|
||||
@@ -39,8 +39,8 @@ use bytes::BytesMut;
|
||||
use futures::prelude::*;
|
||||
use libp2p::core::{upgrade, InboundUpgrade, OutboundUpgrade, UpgradeInfo};
|
||||
use log::{error, warn};
|
||||
use sc_network_common::protocol::ProtocolName;
|
||||
use std::{
|
||||
borrow::Cow,
|
||||
convert::Infallible,
|
||||
io, mem,
|
||||
pin::Pin,
|
||||
@@ -58,7 +58,7 @@ const MAX_HANDSHAKE_SIZE: usize = 1024;
|
||||
pub struct NotificationsIn {
|
||||
/// Protocol name to use when negotiating the substream.
|
||||
/// The first one is the main name, while the other ones are fall backs.
|
||||
protocol_names: Vec<Cow<'static, str>>,
|
||||
protocol_names: Vec<ProtocolName>,
|
||||
/// Maximum allowed size for a single notification.
|
||||
max_notification_size: u64,
|
||||
}
|
||||
@@ -69,7 +69,7 @@ pub struct NotificationsIn {
|
||||
pub struct NotificationsOut {
|
||||
/// Protocol name to use when negotiating the substream.
|
||||
/// The first one is the main name, while the other ones are fall backs.
|
||||
protocol_names: Vec<Cow<'static, str>>,
|
||||
protocol_names: Vec<ProtocolName>,
|
||||
/// Message to send when we start the handshake.
|
||||
initial_message: Vec<u8>,
|
||||
/// Maximum allowed size for a single notification.
|
||||
@@ -114,8 +114,8 @@ pub struct NotificationsOutSubstream<TSubstream> {
|
||||
impl NotificationsIn {
|
||||
/// Builds a new potential upgrade.
|
||||
pub fn new(
|
||||
main_protocol_name: impl Into<Cow<'static, str>>,
|
||||
fallback_names: Vec<Cow<'static, str>>,
|
||||
main_protocol_name: impl Into<ProtocolName>,
|
||||
fallback_names: Vec<ProtocolName>,
|
||||
max_notification_size: u64,
|
||||
) -> Self {
|
||||
let mut protocol_names = fallback_names;
|
||||
@@ -126,16 +126,11 @@ impl NotificationsIn {
|
||||
}
|
||||
|
||||
impl UpgradeInfo for NotificationsIn {
|
||||
type Info = StringProtocolName;
|
||||
type Info = ProtocolName;
|
||||
type InfoIter = vec::IntoIter<Self::Info>;
|
||||
|
||||
fn protocol_info(&self) -> Self::InfoIter {
|
||||
self.protocol_names
|
||||
.iter()
|
||||
.cloned()
|
||||
.map(StringProtocolName)
|
||||
.collect::<Vec<_>>()
|
||||
.into_iter()
|
||||
self.protocol_names.clone().into_iter()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -172,10 +167,10 @@ where
|
||||
|
||||
Ok(NotificationsInOpen {
|
||||
handshake,
|
||||
negotiated_fallback: if negotiated_name.0 == self.protocol_names[0] {
|
||||
negotiated_fallback: if negotiated_name == self.protocol_names[0] {
|
||||
None
|
||||
} else {
|
||||
Some(negotiated_name.0)
|
||||
Some(negotiated_name)
|
||||
},
|
||||
substream,
|
||||
})
|
||||
@@ -189,7 +184,7 @@ pub struct NotificationsInOpen<TSubstream> {
|
||||
pub handshake: Vec<u8>,
|
||||
/// If the negotiated name is not the "main" protocol name but a fallback, contains the
|
||||
/// name of the negotiated fallback.
|
||||
pub negotiated_fallback: Option<Cow<'static, str>>,
|
||||
pub negotiated_fallback: Option<ProtocolName>,
|
||||
/// Implementation of `Stream` that allows receives messages from the substream.
|
||||
pub substream: NotificationsInSubstream<TSubstream>,
|
||||
}
|
||||
@@ -334,8 +329,8 @@ where
|
||||
impl NotificationsOut {
|
||||
/// Builds a new potential upgrade.
|
||||
pub fn new(
|
||||
main_protocol_name: impl Into<Cow<'static, str>>,
|
||||
fallback_names: Vec<Cow<'static, str>>,
|
||||
main_protocol_name: impl Into<ProtocolName>,
|
||||
fallback_names: Vec<ProtocolName>,
|
||||
initial_message: impl Into<Vec<u8>>,
|
||||
max_notification_size: u64,
|
||||
) -> Self {
|
||||
@@ -351,27 +346,12 @@ impl NotificationsOut {
|
||||
}
|
||||
}
|
||||
|
||||
/// Implementation of the `ProtocolName` trait, where the protocol name is a string.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct StringProtocolName(Cow<'static, str>);
|
||||
|
||||
impl upgrade::ProtocolName for StringProtocolName {
|
||||
fn protocol_name(&self) -> &[u8] {
|
||||
self.0.as_bytes()
|
||||
}
|
||||
}
|
||||
|
||||
impl UpgradeInfo for NotificationsOut {
|
||||
type Info = StringProtocolName;
|
||||
type Info = ProtocolName;
|
||||
type InfoIter = vec::IntoIter<Self::Info>;
|
||||
|
||||
fn protocol_info(&self) -> Self::InfoIter {
|
||||
self.protocol_names
|
||||
.iter()
|
||||
.cloned()
|
||||
.map(StringProtocolName)
|
||||
.collect::<Vec<_>>()
|
||||
.into_iter()
|
||||
self.protocol_names.clone().into_iter()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -406,10 +386,10 @@ where
|
||||
|
||||
Ok(NotificationsOutOpen {
|
||||
handshake,
|
||||
negotiated_fallback: if negotiated_name.0 == self.protocol_names[0] {
|
||||
negotiated_fallback: if negotiated_name == self.protocol_names[0] {
|
||||
None
|
||||
} else {
|
||||
Some(negotiated_name.0)
|
||||
Some(negotiated_name)
|
||||
},
|
||||
substream: NotificationsOutSubstream { socket: Framed::new(socket, codec) },
|
||||
})
|
||||
@@ -423,7 +403,7 @@ pub struct NotificationsOutOpen<TSubstream> {
|
||||
pub handshake: Vec<u8>,
|
||||
/// If the negotiated name is not the "main" protocol name but a fallback, contains the
|
||||
/// name of the negotiated fallback.
|
||||
pub negotiated_fallback: Option<Cow<'static, str>>,
|
||||
pub negotiated_fallback: Option<ProtocolName>,
|
||||
/// Implementation of `Sink` that allows sending messages on the substream.
|
||||
pub substream: NotificationsOutSubstream<TSubstream>,
|
||||
}
|
||||
@@ -505,11 +485,10 @@ mod tests {
|
||||
use async_std::net::{TcpListener, TcpStream};
|
||||
use futures::{channel::oneshot, prelude::*};
|
||||
use libp2p::core::upgrade;
|
||||
use std::borrow::Cow;
|
||||
|
||||
#[test]
|
||||
fn basic_works() {
|
||||
const PROTO_NAME: Cow<'static, str> = Cow::Borrowed("/test/proto/1");
|
||||
const PROTO_NAME: &str = "/test/proto/1";
|
||||
let (listener_addr_tx, listener_addr_rx) = oneshot::channel();
|
||||
|
||||
let client = async_std::task::spawn(async move {
|
||||
@@ -552,7 +531,7 @@ mod tests {
|
||||
fn empty_handshake() {
|
||||
// Check that everything still works when the handshake messages are empty.
|
||||
|
||||
const PROTO_NAME: Cow<'static, str> = Cow::Borrowed("/test/proto/1");
|
||||
const PROTO_NAME: &str = "/test/proto/1";
|
||||
let (listener_addr_tx, listener_addr_rx) = oneshot::channel();
|
||||
|
||||
let client = async_std::task::spawn(async move {
|
||||
@@ -593,7 +572,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn refused() {
|
||||
const PROTO_NAME: Cow<'static, str> = Cow::Borrowed("/test/proto/1");
|
||||
const PROTO_NAME: &str = "/test/proto/1";
|
||||
let (listener_addr_tx, listener_addr_rx) = oneshot::channel();
|
||||
|
||||
let client = async_std::task::spawn(async move {
|
||||
@@ -634,7 +613,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn large_initial_message_refused() {
|
||||
const PROTO_NAME: Cow<'static, str> = Cow::Borrowed("/test/proto/1");
|
||||
const PROTO_NAME: &str = "/test/proto/1";
|
||||
let (listener_addr_tx, listener_addr_rx) = oneshot::channel();
|
||||
|
||||
let client = async_std::task::spawn(async move {
|
||||
@@ -672,7 +651,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn large_handshake_refused() {
|
||||
const PROTO_NAME: Cow<'static, str> = Cow::Borrowed("/test/proto/1");
|
||||
const PROTO_NAME: &str = "/test/proto/1";
|
||||
let (listener_addr_tx, listener_addr_rx) = oneshot::channel();
|
||||
|
||||
let client = async_std::task::spawn(async move {
|
||||
|
||||
@@ -50,11 +50,13 @@ use libp2p::{
|
||||
NetworkBehaviourAction, PollParameters,
|
||||
},
|
||||
};
|
||||
use sc_network_common::request_responses::{
|
||||
IfDisconnected, IncomingRequest, OutgoingResponse, ProtocolConfig, RequestFailure,
|
||||
use sc_network_common::{
|
||||
protocol::ProtocolName,
|
||||
request_responses::{
|
||||
IfDisconnected, IncomingRequest, OutgoingResponse, ProtocolConfig, RequestFailure,
|
||||
},
|
||||
};
|
||||
use std::{
|
||||
borrow::Cow,
|
||||
collections::{hash_map::Entry, HashMap},
|
||||
io, iter,
|
||||
pin::Pin,
|
||||
@@ -75,7 +77,7 @@ pub enum Event {
|
||||
/// Peer which has emitted the request.
|
||||
peer: PeerId,
|
||||
/// Name of the protocol in question.
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
/// Whether handling the request was successful or unsuccessful.
|
||||
///
|
||||
/// When successful contains the time elapsed between when we received the request and when
|
||||
@@ -91,7 +93,7 @@ pub enum Event {
|
||||
/// Peer that we send a request to.
|
||||
peer: PeerId,
|
||||
/// Name of the protocol in question.
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
/// Duration the request took.
|
||||
duration: Duration,
|
||||
/// Result of the request.
|
||||
@@ -110,12 +112,12 @@ pub enum Event {
|
||||
/// [`ProtocolRequestId`]s.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
|
||||
struct ProtocolRequestId {
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
request_id: RequestId,
|
||||
}
|
||||
|
||||
impl From<(Cow<'static, str>, RequestId)> for ProtocolRequestId {
|
||||
fn from((protocol, request_id): (Cow<'static, str>, RequestId)) -> Self {
|
||||
impl From<(ProtocolName, RequestId)> for ProtocolRequestId {
|
||||
fn from((protocol, request_id): (ProtocolName, RequestId)) -> Self {
|
||||
Self { protocol, request_id }
|
||||
}
|
||||
}
|
||||
@@ -126,7 +128,7 @@ pub struct RequestResponsesBehaviour {
|
||||
/// Contains the underlying libp2p `RequestResponse` behaviour, plus an optional
|
||||
/// "response builder" used to build responses for incoming requests.
|
||||
protocols: HashMap<
|
||||
Cow<'static, str>,
|
||||
ProtocolName,
|
||||
(RequestResponse<GenericCodec>, Option<mpsc::Sender<IncomingRequest>>),
|
||||
>,
|
||||
|
||||
@@ -162,7 +164,7 @@ struct MessageRequest {
|
||||
request_id: RequestId,
|
||||
request: Vec<u8>,
|
||||
channel: ResponseChannel<Result<Vec<u8>, ()>>,
|
||||
protocol: String,
|
||||
protocol: ProtocolName,
|
||||
resp_builder: Option<futures::channel::mpsc::Sender<IncomingRequest>>,
|
||||
// Once we get incoming request we save all params, create an async call to Peerset
|
||||
// to get the reputation of the peer.
|
||||
@@ -173,7 +175,7 @@ struct MessageRequest {
|
||||
struct RequestProcessingOutcome {
|
||||
peer: PeerId,
|
||||
request_id: RequestId,
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
inner_channel: ResponseChannel<Result<Vec<u8>, ()>>,
|
||||
response: OutgoingResponse,
|
||||
}
|
||||
@@ -495,7 +497,6 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
|
||||
debug_assert!(false, "Received message on outbound-only protocol.");
|
||||
}
|
||||
|
||||
let protocol = Cow::from(protocol);
|
||||
self.pending_responses.push(Box::pin(async move {
|
||||
// The `tx` created above can be dropped if we are not capable of
|
||||
// processing this request, which is reflected as a
|
||||
@@ -618,7 +619,7 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
|
||||
request_id,
|
||||
request,
|
||||
channel,
|
||||
protocol: protocol.to_string(),
|
||||
protocol: protocol.clone(),
|
||||
resp_builder: resp_builder.clone(),
|
||||
get_peer_reputation,
|
||||
});
|
||||
@@ -766,7 +767,7 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
|
||||
pub enum RegisterError {
|
||||
/// A protocol has been specified multiple times.
|
||||
#[error("{0}")]
|
||||
DuplicateProtocol(Cow<'static, str>),
|
||||
DuplicateProtocol(ProtocolName),
|
||||
}
|
||||
|
||||
/// Error when processing a request sent by a remote.
|
||||
|
||||
@@ -61,7 +61,10 @@ use parking_lot::Mutex;
|
||||
use sc_consensus::{BlockImportError, BlockImportStatus, ImportQueue, Link};
|
||||
use sc_network_common::{
|
||||
config::MultiaddrWithPeerId,
|
||||
protocol::event::{DhtEvent, Event},
|
||||
protocol::{
|
||||
event::{DhtEvent, Event},
|
||||
ProtocolName,
|
||||
},
|
||||
request_responses::{IfDisconnected, RequestFailure},
|
||||
service::{
|
||||
NetworkDHTProvider, NetworkEventStream, NetworkNotification, NetworkPeers, NetworkSigner,
|
||||
@@ -76,7 +79,6 @@ use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnbound
|
||||
use sp_blockchain::HeaderBackend;
|
||||
use sp_runtime::traits::{Block as BlockT, NumberFor};
|
||||
use std::{
|
||||
borrow::Cow,
|
||||
cmp,
|
||||
collections::{HashMap, HashSet},
|
||||
fs, iter,
|
||||
@@ -122,7 +124,7 @@ pub struct NetworkService<B: BlockT + 'static, H: ExHashT> {
|
||||
to_worker: TracingUnboundedSender<ServiceToWorkerMsg<B, H>>,
|
||||
/// For each peer and protocol combination, an object that allows sending notifications to
|
||||
/// that peer. Updated by the [`NetworkWorker`].
|
||||
peers_notifications_sinks: Arc<Mutex<HashMap<(PeerId, Cow<'static, str>), NotificationsSink>>>,
|
||||
peers_notifications_sinks: Arc<Mutex<HashMap<(PeerId, ProtocolName), NotificationsSink>>>,
|
||||
/// Field extracted from the [`Metrics`] struct and necessary to report the
|
||||
/// notifications-related metrics.
|
||||
notifications_sizes_metric: Option<HistogramVec>,
|
||||
@@ -890,7 +892,7 @@ where
|
||||
self.peerset.report_peer(who, cost_benefit);
|
||||
}
|
||||
|
||||
fn disconnect_peer(&self, who: PeerId, protocol: Cow<'static, str>) {
|
||||
fn disconnect_peer(&self, who: PeerId, protocol: ProtocolName) {
|
||||
let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::DisconnectPeer(who, protocol));
|
||||
}
|
||||
|
||||
@@ -921,7 +923,7 @@ where
|
||||
|
||||
fn set_reserved_peers(
|
||||
&self,
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
peers: HashSet<Multiaddr>,
|
||||
) -> Result<(), String> {
|
||||
let peers_addrs = self.split_multiaddr_and_peer_id(peers)?;
|
||||
@@ -952,7 +954,7 @@ where
|
||||
|
||||
fn add_peers_to_reserved_set(
|
||||
&self,
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
peers: HashSet<Multiaddr>,
|
||||
) -> Result<(), String> {
|
||||
let peers = self.split_multiaddr_and_peer_id(peers)?;
|
||||
@@ -976,7 +978,7 @@ where
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn remove_peers_from_reserved_set(&self, protocol: Cow<'static, str>, peers: Vec<PeerId>) {
|
||||
fn remove_peers_from_reserved_set(&self, protocol: ProtocolName, peers: Vec<PeerId>) {
|
||||
for peer_id in peers.into_iter() {
|
||||
let _ = self
|
||||
.to_worker
|
||||
@@ -986,7 +988,7 @@ where
|
||||
|
||||
fn add_to_peers_set(
|
||||
&self,
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
peers: HashSet<Multiaddr>,
|
||||
) -> Result<(), String> {
|
||||
let peers = self.split_multiaddr_and_peer_id(peers)?;
|
||||
@@ -1010,7 +1012,7 @@ where
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn remove_from_peers_set(&self, protocol: Cow<'static, str>, peers: Vec<PeerId>) {
|
||||
fn remove_from_peers_set(&self, protocol: ProtocolName, peers: Vec<PeerId>) {
|
||||
for peer_id in peers.into_iter() {
|
||||
let _ = self
|
||||
.to_worker
|
||||
@@ -1040,7 +1042,7 @@ where
|
||||
B: BlockT + 'static,
|
||||
H: ExHashT,
|
||||
{
|
||||
fn write_notification(&self, target: PeerId, protocol: Cow<'static, str>, message: Vec<u8>) {
|
||||
fn write_notification(&self, target: PeerId, protocol: ProtocolName, message: Vec<u8>) {
|
||||
// We clone the `NotificationsSink` in order to be able to unlock the network-wide
|
||||
// `peers_notifications_sinks` mutex as soon as possible.
|
||||
let sink = {
|
||||
@@ -1077,7 +1079,7 @@ where
|
||||
fn notification_sender(
|
||||
&self,
|
||||
target: PeerId,
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
) -> Result<Box<dyn NotificationSenderT>, NotificationSenderError> {
|
||||
// We clone the `NotificationsSink` in order to be able to unlock the network-wide
|
||||
// `peers_notifications_sinks` mutex as soon as possible.
|
||||
@@ -1108,7 +1110,7 @@ where
|
||||
async fn request(
|
||||
&self,
|
||||
target: PeerId,
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
request: Vec<u8>,
|
||||
connect: IfDisconnected,
|
||||
) -> Result<Vec<u8>, RequestFailure> {
|
||||
@@ -1128,7 +1130,7 @@ where
|
||||
fn start_request(
|
||||
&self,
|
||||
target: PeerId,
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
request: Vec<u8>,
|
||||
tx: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
|
||||
connect: IfDisconnected,
|
||||
@@ -1179,7 +1181,7 @@ pub struct NotificationSender {
|
||||
sink: NotificationsSink,
|
||||
|
||||
/// Name of the protocol on the wire.
|
||||
protocol_name: Cow<'static, str>,
|
||||
protocol_name: ProtocolName,
|
||||
|
||||
/// Field extracted from the [`Metrics`] struct and necessary to report the
|
||||
/// notifications-related metrics.
|
||||
@@ -1212,7 +1214,7 @@ pub struct NotificationSenderReady<'a> {
|
||||
peer_id: &'a PeerId,
|
||||
|
||||
/// Name of the protocol on the wire.
|
||||
protocol_name: &'a Cow<'static, str>,
|
||||
protocol_name: &'a ProtocolName,
|
||||
|
||||
/// Field extracted from the [`Metrics`] struct and necessary to report the
|
||||
/// notifications-related metrics.
|
||||
@@ -1256,16 +1258,16 @@ enum ServiceToWorkerMsg<B: BlockT, H: ExHashT> {
|
||||
AddReserved(PeerId),
|
||||
RemoveReserved(PeerId),
|
||||
SetReserved(HashSet<PeerId>),
|
||||
SetPeersetReserved(Cow<'static, str>, HashSet<PeerId>),
|
||||
AddSetReserved(Cow<'static, str>, PeerId),
|
||||
RemoveSetReserved(Cow<'static, str>, PeerId),
|
||||
AddToPeersSet(Cow<'static, str>, PeerId),
|
||||
RemoveFromPeersSet(Cow<'static, str>, PeerId),
|
||||
SetPeersetReserved(ProtocolName, HashSet<PeerId>),
|
||||
AddSetReserved(ProtocolName, PeerId),
|
||||
RemoveSetReserved(ProtocolName, PeerId),
|
||||
AddToPeersSet(ProtocolName, PeerId),
|
||||
RemoveFromPeersSet(ProtocolName, PeerId),
|
||||
SyncFork(Vec<PeerId>, B::Hash, NumberFor<B>),
|
||||
EventStream(out_events::Sender),
|
||||
Request {
|
||||
target: PeerId,
|
||||
protocol: Cow<'static, str>,
|
||||
protocol: ProtocolName,
|
||||
request: Vec<u8>,
|
||||
pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
|
||||
connect: IfDisconnected,
|
||||
@@ -1276,7 +1278,7 @@ enum ServiceToWorkerMsg<B: BlockT, H: ExHashT> {
|
||||
NetworkState {
|
||||
pending_response: oneshot::Sender<Result<NetworkState, RequestFailure>>,
|
||||
},
|
||||
DisconnectPeer(PeerId, Cow<'static, str>),
|
||||
DisconnectPeer(PeerId, ProtocolName),
|
||||
NewBestBlockImported(B::Hash, NumberFor<B>),
|
||||
}
|
||||
|
||||
@@ -1312,7 +1314,7 @@ where
|
||||
boot_node_ids: Arc<HashSet<PeerId>>,
|
||||
/// For each peer and protocol combination, an object that allows sending notifications to
|
||||
/// that peer. Shared with the [`NetworkService`].
|
||||
peers_notifications_sinks: Arc<Mutex<HashMap<(PeerId, Cow<'static, str>), NotificationsSink>>>,
|
||||
peers_notifications_sinks: Arc<Mutex<HashMap<(PeerId, ProtocolName), NotificationsSink>>>,
|
||||
/// Controller for the handler of incoming and outgoing transactions.
|
||||
tx_handler_controller: transactions::TransactionsHandlerController<H>,
|
||||
}
|
||||
@@ -1456,7 +1458,7 @@ where
|
||||
.network_service
|
||||
.behaviour_mut()
|
||||
.user_protocol_mut()
|
||||
.disconnect_peer(&who, &protocol_name),
|
||||
.disconnect_peer(&who, protocol_name),
|
||||
ServiceToWorkerMsg::NewBestBlockImported(hash, number) => this
|
||||
.network_service
|
||||
.behaviour_mut()
|
||||
|
||||
@@ -32,7 +32,7 @@ use sc_network_sync::{
|
||||
};
|
||||
use sp_consensus::block_validation::DefaultBlockAnnounceValidator;
|
||||
use sp_runtime::traits::{Block as BlockT, Header as _};
|
||||
use std::{borrow::Cow, sync::Arc, time::Duration};
|
||||
use std::{sync::Arc, time::Duration};
|
||||
use substrate_test_runtime_client::{TestClientBuilder, TestClientBuilderExt as _};
|
||||
|
||||
type TestNetworkService = NetworkService<
|
||||
@@ -165,7 +165,7 @@ fn build_test_full_node(
|
||||
(service, event_stream)
|
||||
}
|
||||
|
||||
const PROTOCOL_NAME: Cow<'static, str> = Cow::Borrowed("/foo");
|
||||
const PROTOCOL_NAME: &str = "/foo";
|
||||
|
||||
/// Builds two nodes and their associated events stream.
|
||||
/// The nodes are connected together and have the `PROTOCOL_NAME` protocol registered.
|
||||
@@ -179,7 +179,7 @@ fn build_nodes_one_proto() -> (
|
||||
|
||||
let (node1, events_stream1) = build_test_full_node(config::NetworkConfiguration {
|
||||
extra_sets: vec![config::NonDefaultSetConfig {
|
||||
notifications_protocol: PROTOCOL_NAME,
|
||||
notifications_protocol: PROTOCOL_NAME.into(),
|
||||
fallback_names: Vec::new(),
|
||||
max_notification_size: 1024 * 1024,
|
||||
set_config: Default::default(),
|
||||
@@ -191,7 +191,7 @@ fn build_nodes_one_proto() -> (
|
||||
|
||||
let (node2, events_stream2) = build_test_full_node(config::NetworkConfiguration {
|
||||
extra_sets: vec![config::NonDefaultSetConfig {
|
||||
notifications_protocol: PROTOCOL_NAME,
|
||||
notifications_protocol: PROTOCOL_NAME.into(),
|
||||
fallback_names: Vec::new(),
|
||||
max_notification_size: 1024 * 1024,
|
||||
set_config: config::SetConfig {
|
||||
@@ -219,10 +219,18 @@ fn notifications_state_consistent() {
|
||||
|
||||
// Write some initial notifications that shouldn't get through.
|
||||
for _ in 0..(rand::random::<u8>() % 5) {
|
||||
node1.write_notification(node2.local_peer_id(), PROTOCOL_NAME, b"hello world".to_vec());
|
||||
node1.write_notification(
|
||||
node2.local_peer_id(),
|
||||
PROTOCOL_NAME.into(),
|
||||
b"hello world".to_vec(),
|
||||
);
|
||||
}
|
||||
for _ in 0..(rand::random::<u8>() % 5) {
|
||||
node2.write_notification(node1.local_peer_id(), PROTOCOL_NAME, b"hello world".to_vec());
|
||||
node2.write_notification(
|
||||
node1.local_peer_id(),
|
||||
PROTOCOL_NAME.into(),
|
||||
b"hello world".to_vec(),
|
||||
);
|
||||
}
|
||||
|
||||
async_std::task::block_on(async move {
|
||||
@@ -247,24 +255,24 @@ fn notifications_state_consistent() {
|
||||
if rand::random::<u8>() % 5 >= 3 {
|
||||
node1.write_notification(
|
||||
node2.local_peer_id(),
|
||||
PROTOCOL_NAME,
|
||||
PROTOCOL_NAME.into(),
|
||||
b"hello world".to_vec(),
|
||||
);
|
||||
}
|
||||
if rand::random::<u8>() % 5 >= 3 {
|
||||
node2.write_notification(
|
||||
node1.local_peer_id(),
|
||||
PROTOCOL_NAME,
|
||||
PROTOCOL_NAME.into(),
|
||||
b"hello world".to_vec(),
|
||||
);
|
||||
}
|
||||
|
||||
// Also randomly disconnect the two nodes from time to time.
|
||||
if rand::random::<u8>() % 20 == 0 {
|
||||
node1.disconnect_peer(node2.local_peer_id(), PROTOCOL_NAME);
|
||||
node1.disconnect_peer(node2.local_peer_id(), PROTOCOL_NAME.into());
|
||||
}
|
||||
if rand::random::<u8>() % 20 == 0 {
|
||||
node2.disconnect_peer(node1.local_peer_id(), PROTOCOL_NAME);
|
||||
node2.disconnect_peer(node1.local_peer_id(), PROTOCOL_NAME.into());
|
||||
}
|
||||
|
||||
// Grab next event from either `events_stream1` or `events_stream2`.
|
||||
@@ -288,7 +296,7 @@ fn notifications_state_consistent() {
|
||||
future::Either::Left(Event::NotificationStreamOpened {
|
||||
remote, protocol, ..
|
||||
}) =>
|
||||
if protocol == PROTOCOL_NAME {
|
||||
if protocol == PROTOCOL_NAME.into() {
|
||||
something_happened = true;
|
||||
assert!(!node1_to_node2_open);
|
||||
node1_to_node2_open = true;
|
||||
@@ -297,7 +305,7 @@ fn notifications_state_consistent() {
|
||||
future::Either::Right(Event::NotificationStreamOpened {
|
||||
remote, protocol, ..
|
||||
}) =>
|
||||
if protocol == PROTOCOL_NAME {
|
||||
if protocol == PROTOCOL_NAME.into() {
|
||||
something_happened = true;
|
||||
assert!(!node2_to_node1_open);
|
||||
node2_to_node1_open = true;
|
||||
@@ -306,7 +314,7 @@ fn notifications_state_consistent() {
|
||||
future::Either::Left(Event::NotificationStreamClosed {
|
||||
remote, protocol, ..
|
||||
}) =>
|
||||
if protocol == PROTOCOL_NAME {
|
||||
if protocol == PROTOCOL_NAME.into() {
|
||||
assert!(node1_to_node2_open);
|
||||
node1_to_node2_open = false;
|
||||
assert_eq!(remote, node2.local_peer_id());
|
||||
@@ -314,7 +322,7 @@ fn notifications_state_consistent() {
|
||||
future::Either::Right(Event::NotificationStreamClosed {
|
||||
remote, protocol, ..
|
||||
}) =>
|
||||
if protocol == PROTOCOL_NAME {
|
||||
if protocol == PROTOCOL_NAME.into() {
|
||||
assert!(node2_to_node1_open);
|
||||
node2_to_node1_open = false;
|
||||
assert_eq!(remote, node1.local_peer_id());
|
||||
@@ -325,7 +333,7 @@ fn notifications_state_consistent() {
|
||||
if rand::random::<u8>() % 5 >= 4 {
|
||||
node1.write_notification(
|
||||
node2.local_peer_id(),
|
||||
PROTOCOL_NAME,
|
||||
PROTOCOL_NAME.into(),
|
||||
b"hello world".to_vec(),
|
||||
);
|
||||
}
|
||||
@@ -336,7 +344,7 @@ fn notifications_state_consistent() {
|
||||
if rand::random::<u8>() % 5 >= 4 {
|
||||
node2.write_notification(
|
||||
node1.local_peer_id(),
|
||||
PROTOCOL_NAME,
|
||||
PROTOCOL_NAME.into(),
|
||||
b"hello world".to_vec(),
|
||||
);
|
||||
}
|
||||
@@ -361,7 +369,7 @@ fn lots_of_incoming_peers_works() {
|
||||
let (main_node, _) = build_test_full_node(config::NetworkConfiguration {
|
||||
listen_addresses: vec![listen_addr.clone()],
|
||||
extra_sets: vec![config::NonDefaultSetConfig {
|
||||
notifications_protocol: PROTOCOL_NAME,
|
||||
notifications_protocol: PROTOCOL_NAME.into(),
|
||||
fallback_names: Vec::new(),
|
||||
max_notification_size: 1024 * 1024,
|
||||
set_config: config::SetConfig { in_peers: u32::MAX, ..Default::default() },
|
||||
@@ -380,7 +388,7 @@ fn lots_of_incoming_peers_works() {
|
||||
let (_dialing_node, event_stream) = build_test_full_node(config::NetworkConfiguration {
|
||||
listen_addresses: vec![],
|
||||
extra_sets: vec![config::NonDefaultSetConfig {
|
||||
notifications_protocol: PROTOCOL_NAME,
|
||||
notifications_protocol: PROTOCOL_NAME.into(),
|
||||
fallback_names: Vec::new(),
|
||||
max_notification_size: 1024 * 1024,
|
||||
set_config: config::SetConfig {
|
||||
@@ -449,7 +457,7 @@ fn notifications_back_pressure() {
|
||||
Event::NotificationStreamClosed { .. } => panic!(),
|
||||
Event::NotificationsReceived { messages, .. } =>
|
||||
for message in messages {
|
||||
assert_eq!(message.0, PROTOCOL_NAME);
|
||||
assert_eq!(message.0, PROTOCOL_NAME.into());
|
||||
assert_eq!(message.1, format!("hello #{}", received_notifications));
|
||||
received_notifications += 1;
|
||||
},
|
||||
@@ -473,7 +481,7 @@ fn notifications_back_pressure() {
|
||||
|
||||
// Sending!
|
||||
for num in 0..TOTAL_NOTIFS {
|
||||
let notif = node1.notification_sender(node2_id, PROTOCOL_NAME).unwrap();
|
||||
let notif = node1.notification_sender(node2_id, PROTOCOL_NAME.into()).unwrap();
|
||||
notif
|
||||
.ready()
|
||||
.await
|
||||
@@ -491,15 +499,14 @@ fn fallback_name_working() {
|
||||
// Node 1 supports the protocols "new" and "old". Node 2 only supports "old". Checks whether
|
||||
// they can connect.
|
||||
|
||||
const NEW_PROTOCOL_NAME: Cow<'static, str> =
|
||||
Cow::Borrowed("/new-shiny-protocol-that-isnt-PROTOCOL_NAME");
|
||||
const NEW_PROTOCOL_NAME: &str = "/new-shiny-protocol-that-isnt-PROTOCOL_NAME";
|
||||
|
||||
let listen_addr = config::build_multiaddr![Memory(rand::random::<u64>())];
|
||||
|
||||
let (node1, mut events_stream1) = build_test_full_node(config::NetworkConfiguration {
|
||||
extra_sets: vec![config::NonDefaultSetConfig {
|
||||
notifications_protocol: NEW_PROTOCOL_NAME.clone(),
|
||||
fallback_names: vec![PROTOCOL_NAME],
|
||||
notifications_protocol: NEW_PROTOCOL_NAME.into(),
|
||||
fallback_names: vec![PROTOCOL_NAME.into()],
|
||||
max_notification_size: 1024 * 1024,
|
||||
set_config: Default::default(),
|
||||
}],
|
||||
@@ -510,7 +517,7 @@ fn fallback_name_working() {
|
||||
|
||||
let (_, mut events_stream2) = build_test_full_node(config::NetworkConfiguration {
|
||||
extra_sets: vec![config::NonDefaultSetConfig {
|
||||
notifications_protocol: PROTOCOL_NAME,
|
||||
notifications_protocol: PROTOCOL_NAME.into(),
|
||||
fallback_names: Vec::new(),
|
||||
max_notification_size: 1024 * 1024,
|
||||
set_config: config::SetConfig {
|
||||
@@ -531,7 +538,7 @@ fn fallback_name_working() {
|
||||
loop {
|
||||
match events_stream2.next().await.unwrap() {
|
||||
Event::NotificationStreamOpened { protocol, negotiated_fallback, .. } => {
|
||||
assert_eq!(protocol, PROTOCOL_NAME);
|
||||
assert_eq!(protocol, PROTOCOL_NAME.into());
|
||||
assert_eq!(negotiated_fallback, None);
|
||||
break
|
||||
},
|
||||
@@ -545,9 +552,9 @@ fn fallback_name_working() {
|
||||
loop {
|
||||
match events_stream1.next().await.unwrap() {
|
||||
Event::NotificationStreamOpened { protocol, negotiated_fallback, .. }
|
||||
if protocol == NEW_PROTOCOL_NAME =>
|
||||
if protocol == NEW_PROTOCOL_NAME.into() =>
|
||||
{
|
||||
assert_eq!(negotiated_fallback, Some(PROTOCOL_NAME));
|
||||
assert_eq!(negotiated_fallback, Some(PROTOCOL_NAME.into()));
|
||||
break
|
||||
},
|
||||
_ => {},
|
||||
|
||||
@@ -42,12 +42,14 @@ use log::{debug, trace, warn};
|
||||
use prometheus_endpoint::{register, Counter, PrometheusError, Registry, U64};
|
||||
use sc_network_common::{
|
||||
config::ProtocolId,
|
||||
protocol::event::{Event, ObservedRole},
|
||||
protocol::{
|
||||
event::{Event, ObservedRole},
|
||||
ProtocolName,
|
||||
},
|
||||
service::{NetworkEventStream, NetworkNotification, NetworkPeers},
|
||||
};
|
||||
use sp_runtime::traits::Block as BlockT;
|
||||
use std::{
|
||||
borrow::Cow,
|
||||
collections::{hash_map::Entry, HashMap},
|
||||
iter,
|
||||
num::NonZeroUsize,
|
||||
@@ -130,8 +132,8 @@ impl<H: ExHashT> Future for PendingTransaction<H> {
|
||||
|
||||
/// Prototype for a [`TransactionsHandler`].
|
||||
pub struct TransactionsHandlerPrototype {
|
||||
protocol_name: Cow<'static, str>,
|
||||
fallback_protocol_names: Vec<Cow<'static, str>>,
|
||||
protocol_name: ProtocolName,
|
||||
fallback_protocol_names: Vec<ProtocolName>,
|
||||
}
|
||||
|
||||
impl TransactionsHandlerPrototype {
|
||||
@@ -244,7 +246,7 @@ enum ToHandler<H: ExHashT> {
|
||||
|
||||
/// Handler for transactions. Call [`TransactionsHandler::run`] to start the processing.
|
||||
pub struct TransactionsHandler<B: BlockT + 'static, H: ExHashT> {
|
||||
protocol_name: Cow<'static, str>,
|
||||
protocol_name: ProtocolName,
|
||||
/// Interval at which we call `propagate_transactions`.
|
||||
propagate_timeout: Pin<Box<dyn Stream<Item = ()> + Send>>,
|
||||
/// Pending transactions verification tasks.
|
||||
|
||||
Reference in New Issue
Block a user