feat: integrate multiplexing (#5559)

crates/net/eth-wire/Cargo.toml

@@ -53,6 +53,7 @@ secp256k1 = { workspace = true, features = ["global-context", "rand-std", "recov
 arbitrary = { workspace = true, features = ["derive"] }
 proptest.workspace = true
 proptest-derive.workspace = true
+async-stream = "0.3"
 
 [features]
 default = ["serde"]

crates/net/eth-wire/src/capability.rs

@@ -317,6 +317,14 @@ impl SharedCapability {
         }
     }
 
+    /// Returns the eth version if it's the `eth` capability.
+    pub fn eth_version(&self) -> Option<EthVersion> {
+        match self {
+            SharedCapability::Eth { version, .. } => Some(*version),
+            _ => None,
+        }
+    }
+
     /// Returns the message ID offset of the current capability.
     ///
     /// This represents the message ID offset for the first message of the eth capability in the
@@ -375,8 +383,8 @@ impl SharedCapabilities {
 
     /// Returns the negotiated eth version if it is shared.
     #[inline]
-    pub fn eth_version(&self) -> Result<u8, P2PStreamError> {
-        self.eth().map(|cap| cap.version())
+    pub fn eth_version(&self) -> Result<EthVersion, P2PStreamError> {
+        self.eth().map(|cap| cap.eth_version().expect("is eth; qed"))
     }
 
     /// Returns true if the shared capabilities contain the given capability.
@@ -438,6 +446,18 @@ impl SharedCapabilities {
     ) -> Result<&SharedCapability, UnsupportedCapabilityError> {
         self.find(cap).ok_or_else(|| UnsupportedCapabilityError { capability: cap.clone() })
     }
+
+    /// Returns the number of shared capabilities.
+    #[inline]
+    pub fn len(&self) -> usize {
+        self.0.len()
+    }
+
+    /// Returns true if there are no shared capabilities.
+    #[inline]
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
 }
 
 /// Determines the offsets for each shared capability between the input list of peer

crates/net/eth-wire/src/ethstream.rs

@@ -166,6 +166,7 @@ where
 #[pin_project]
 #[derive(Debug)]
 pub struct EthStream<S> {
+    /// Negotiated eth version.
     version: EthVersion,
     #[pin]
     inner: S,
@@ -174,26 +175,31 @@ pub struct EthStream<S> {
 impl<S> EthStream<S> {
     /// Creates a new unauthed [`EthStream`] from a provided stream. You will need
     /// to manually handshake a peer.
+    #[inline]
     pub fn new(version: EthVersion, inner: S) -> Self {
         Self { version, inner }
     }
 
     /// Returns the eth version.
+    #[inline]
     pub fn version(&self) -> EthVersion {
         self.version
     }
 
     /// Returns the underlying stream.
+    #[inline]
     pub fn inner(&self) -> &S {
         &self.inner
     }
 
     /// Returns mutable access to the underlying stream.
+    #[inline]
     pub fn inner_mut(&mut self) -> &mut S {
         &mut self.inner
     }
 
     /// Consumes this type and returns the wrapped stream.
+    #[inline]
     pub fn into_inner(self) -> S {
         self.inner
     }

crates/net/eth-wire/src/hello.rs

@@ -49,6 +49,7 @@ impl HelloMessageWithProtocols {
     }
 
     /// Returns the raw [HelloMessage] without the additional protocol information.
+    #[inline]
     pub fn message(&self) -> HelloMessage {
         HelloMessage {
             protocol_version: self.protocol_version,
@@ -69,6 +70,25 @@ impl HelloMessageWithProtocols {
             id: self.id,
         }
     }
+
+    /// Returns true if the set of protocols contains the given protocol.
+    #[inline]
+    pub fn contains_protocol(&self, protocol: &Protocol) -> bool {
+        self.protocols.iter().any(|p| p.cap == protocol.cap)
+    }
+
+    /// Adds a new protocol to the set.
+    ///
+    /// Returns an error if the protocol already exists.
+    #[inline]
+    pub fn try_add_protocol(&mut self, protocol: Protocol) -> Result<(), Protocol> {
+        if self.contains_protocol(&protocol) {
+            Err(protocol)
+        } else {
+            self.protocols.push(protocol);
+            Ok(())
+        }
+    }
 }
 
 // TODO: determine if we should allow for the extra fields at the end like EIP-706 suggests