draft: progressing on removing {In,Out}boundUpgrade

core/CHANGELOG.md

@@ -2,6 +2,8 @@
 
 - Implement `{In,Out}boundConnectionUpgrade` for `SelectUpgrade`.
   See [PR 4812](https://github.com/libp2p/rust-libp2p/pull/4812).
+- Deprecate "upgrade" implementations in preparation for removal.
+  See [PR XXXX](https://github.com/libp2p/rust-libp2p/pull/XXXX).
 
 ## 0.41.0
 

core/src/upgrade.rs

@@ -58,10 +58,8 @@
 //!
 
 mod apply;
-mod denied;
 mod either;
 mod error;
-mod pending;
 mod ready;
 mod select;
 
@@ -71,9 +69,8 @@ pub(crate) use apply::{
 pub(crate) use error::UpgradeError;
 use futures::future::Future;
 
-pub use self::{
-    denied::DeniedUpgrade, pending::PendingUpgrade, ready::ReadyUpgrade, select::SelectUpgrade,
-};
+#[allow(deprecated)]
+pub use self::{ready::ReadyUpgrade, select::SelectUpgrade};
 pub use crate::Negotiated;
 pub use multistream_select::{NegotiatedComplete, NegotiationError, ProtocolError, Version};
 

core/src/upgrade/ready.rs

@@ -19,13 +19,16 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
+#![allow(deprecated)]
+
 use crate::upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo};
 use futures::future;
 use std::iter;
 use void::Void;
 
 /// Implementation of [`UpgradeInfo`], [`InboundUpgrade`] and [`OutboundUpgrade`] that directly yields the substream.
 #[derive(Debug, Copy, Clone)]
+#[deprecated(note = "Use `libp2p::swarm::SingleProtocol` instead.")]
 pub struct ReadyUpgrade<P> {
     protocol_name: P,
 }

protocols/autonat/src/protocol.rs

@@ -37,7 +37,6 @@ pub struct AutoNatCodec;
 
 #[async_trait]
 impl request_response::Codec for AutoNatCodec {
-    type Protocol = StreamProtocol;
     type Request = DialRequest;
     type Response = DialResponse;
 

protocols/dcutr/src/handler/relayed.rs

@@ -25,14 +25,13 @@ use crate::{protocol, PROTOCOL_NAME};
 use either::Either;
 use futures::future;
 use libp2p_core::multiaddr::Multiaddr;
-use libp2p_core::upgrade::{DeniedUpgrade, ReadyUpgrade};
 use libp2p_core::ConnectedPoint;
 use libp2p_swarm::handler::{
     ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
     ListenUpgradeError,
 };
 use libp2p_swarm::{
-    ConnectionHandler, ConnectionHandlerEvent, StreamProtocol, StreamUpgradeError,
+    ConnectionHandler, ConnectionHandlerEvent, NoProtocols, SingleProtocol, StreamUpgradeError,
     SubstreamProtocol,
 };
 use protocol::{inbound, outbound};
@@ -180,23 +179,23 @@ impl Handler {
 impl ConnectionHandler for Handler {
     type FromBehaviour = Command;
     type ToBehaviour = Event;
-    type InboundProtocol = Either<ReadyUpgrade<StreamProtocol>, DeniedUpgrade>;
-    type OutboundProtocol = ReadyUpgrade<StreamProtocol>;
+    type InboundProtocol = Either<SingleProtocol, NoProtocols>;
+    type OutboundProtocol = SingleProtocol;
     type OutboundOpenInfo = ();
     type InboundOpenInfo = ();
 
     fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
         match self.endpoint {
             ConnectedPoint::Dialer { .. } => {
-                SubstreamProtocol::new(Either::Left(ReadyUpgrade::new(PROTOCOL_NAME)), ())
+                SubstreamProtocol::new(Either::Left(SingleProtocol::new(PROTOCOL_NAME)), ())
             }
             ConnectedPoint::Listener { .. } => {
                 // By the protocol specification the listening side of a relayed connection
                 // initiates the _direct connection upgrade_. In other words the listening side of
                 // the relayed connection opens a substream to the dialing side. (Connection roles
                 // and substream roles are reversed.) The listening side on a relayed connection
                 // never expects incoming substreams, hence the denied upgrade below.
-                SubstreamProtocol::new(Either::Right(DeniedUpgrade), ())
+                SubstreamProtocol::new(Either::Right(NoProtocols::new()), ())
             }
         }
     }
@@ -206,7 +205,7 @@ impl ConnectionHandler for Handler {
             Command::Connect => {
                 self.queued_events
                     .push_back(ConnectionHandlerEvent::OutboundSubstreamRequest {
-                        protocol: SubstreamProtocol::new(ReadyUpgrade::new(PROTOCOL_NAME), ()),
+                        protocol: SubstreamProtocol::new(SingleProtocol::new(PROTOCOL_NAME), ()),
                     });
                 self.attempts += 1;
             }

protocols/gossipsub/src/handler.rs

@@ -27,12 +27,11 @@ use futures::future::Either;
 use futures::prelude::*;
 use futures::StreamExt;
 use instant::Instant;
-use libp2p_core::upgrade::DeniedUpgrade;
 use libp2p_swarm::handler::{
     ConnectionEvent, ConnectionHandler, ConnectionHandlerEvent, DialUpgradeError,
     FullyNegotiatedInbound, FullyNegotiatedOutbound, StreamUpgradeError, SubstreamProtocol,
 };
-use libp2p_swarm::Stream;
+use libp2p_swarm::{NoProtocols, Stream};
 use smallvec::SmallVec;
 use std::{
     pin::Pin,
@@ -390,7 +389,7 @@ impl ConnectionHandler for Handler {
     type FromBehaviour = HandlerIn;
     type ToBehaviour = HandlerEvent;
     type InboundOpenInfo = ();
-    type InboundProtocol = either::Either<ProtocolConfig, DeniedUpgrade>;
+    type InboundProtocol = either::Either<ProtocolConfig, NoProtocols>;
     type OutboundOpenInfo = ();
     type OutboundProtocol = ProtocolConfig;
 
@@ -400,7 +399,7 @@ impl ConnectionHandler for Handler {
                 SubstreamProtocol::new(either::Either::Left(handler.listen_protocol.clone()), ())
             }
             Handler::Disabled(_) => {
-                SubstreamProtocol::new(either::Either::Right(DeniedUpgrade), ())
+                SubstreamProtocol::new(either::Either::Right(NoProtocols::new()), ())
             }
         }
     }

protocols/gossipsub/src/protocol.rs

@@ -90,6 +90,12 @@ impl AsRef<str> for ProtocolId {
     }
 }
 
+impl AsRef<StreamProtocol> for ProtocolId {
+    fn as_ref(&self) -> &StreamProtocol {
+        self.protocol.as_ref()
+    }
+}
+
 impl UpgradeInfo for ProtocolConfig {
     type Info = ProtocolId;
     type InfoIter = Vec<Self::Info>;

protocols/kad/src/behaviour.rs

@@ -26,7 +26,7 @@ use crate::addresses::Addresses;
 use crate::handler::{Handler, HandlerEvent, HandlerIn, RequestId};
 use crate::jobs::*;
 use crate::kbucket::{self, Distance, KBucketsTable, NodeStatus};
-use crate::protocol::{ConnectionType, KadPeer, ProtocolConfig};
+use crate::protocol::{ConnectionType, KadPeer, DEFAULT_MAX_PACKET_SIZE, DEFAULT_PROTO_NAME};
 use crate::query::{Query, QueryConfig, QueryId, QueryPool, QueryPoolState};
 use crate::record::{
     self,
@@ -68,8 +68,8 @@ pub struct Behaviour<TStore> {
     /// The k-bucket insertion strategy.
     kbucket_inserts: BucketInserts,
 
-    /// Configuration of the wire protocol.
-    protocol_config: ProtocolConfig,
+    protocol_names: Vec<StreamProtocol>,
+    max_packet_size: usize,
 
     /// Configuration of [`RecordStore`] filtering.
     record_filtering: StoreInserts,
@@ -172,7 +172,8 @@ pub enum StoreInserts {
 pub struct Config {
     kbucket_pending_timeout: Duration,
     query_config: QueryConfig,
-    protocol_config: ProtocolConfig,
+    protocol_names: Vec<StreamProtocol>,
+    max_packet_size: usize,
     record_ttl: Option<Duration>,
     record_replication_interval: Option<Duration>,
     record_publication_interval: Option<Duration>,
@@ -188,7 +189,7 @@ impl Default for Config {
         Config {
             kbucket_pending_timeout: Duration::from_secs(60),
             query_config: QueryConfig::default(),
-            protocol_config: Default::default(),
+            protocol_names: vec![DEFAULT_PROTO_NAME],
             record_ttl: Some(Duration::from_secs(36 * 60 * 60)),
             record_replication_interval: Some(Duration::from_secs(60 * 60)),
             record_publication_interval: Some(Duration::from_secs(24 * 60 * 60)),
@@ -197,6 +198,7 @@ impl Default for Config {
             provider_record_ttl: Some(Duration::from_secs(24 * 60 * 60)),
             kbucket_inserts: BucketInserts::OnConnected,
             caching: Caching::Enabled { max_peers: 1 },
+            max_packet_size: DEFAULT_MAX_PACKET_SIZE,
         }
     }
 }
@@ -227,7 +229,7 @@ impl Config {
     /// be able to talk to other nodes supporting any of the provided names.
     /// Multiple names must be used with caution to avoid network partitioning.
     pub fn set_protocol_names(&mut self, names: Vec<StreamProtocol>) -> &mut Self {
-        self.protocol_config.set_protocol_names(names);
+        self.protocol_names = names;
         self
     }
 
@@ -371,7 +373,7 @@ impl Config {
     /// It might be necessary to increase this value if trying to put large
     /// records.
     pub fn set_max_packet_size(&mut self, size: usize) -> &mut Self {
-        self.protocol_config.set_max_packet_size(size);
+        self.max_packet_size = size;
         self
     }
 
@@ -404,7 +406,7 @@ where
 
     /// Get the protocol name of this kademlia instance.
     pub fn protocol_names(&self) -> &[StreamProtocol] {
-        self.protocol_config.protocol_names()
+        self.protocol_names.as_slice()
     }
 
     /// Creates a new `Kademlia` network behaviour with the given configuration.
@@ -432,7 +434,8 @@ where
             caching: config.caching,
             kbuckets: KBucketsTable::new(local_key, config.kbucket_pending_timeout),
             kbucket_inserts: config.kbucket_inserts,
-            protocol_config: config.protocol_config,
+            protocol_names: config.protocol_names,
+            max_packet_size: config.max_packet_size,
             record_filtering: config.record_filtering,
             queued_events: VecDeque::with_capacity(config.query_config.replication_factor.get()),
             listen_addresses: Default::default(),
@@ -2082,7 +2085,8 @@ where
         };
 
         let mut handler = Handler::new(
-            self.protocol_config.clone(),
+            self.protocol_names.clone(),
+            self.max_packet_size,
             connected_point,
             peer,
             self.mode,
@@ -2105,7 +2109,8 @@ where
         };
 
         let mut handler = Handler::new(
-            self.protocol_config.clone(),
+            self.protocol_names.clone(),
+            self.max_packet_size,
             connected_point,
             peer,
             self.mode,