chore(*): merge latest changes

Cargo.toml

@@ -108,7 +108,7 @@ smallvec = "1.6.1"
 libp2p-deflate = { version = "0.35.0", path = "transports/deflate", optional = true }
 libp2p-dns = { version = "0.35.0", path = "transports/dns", optional = true, default-features = false }
 libp2p-mdns = { version = "0.39.0", path = "protocols/mdns", optional = true, default-features = false }
-libp2p-tcp = { version = "0.34.0", path = "transports/tcp", default-features = false, optional = true }
+libp2p-tcp = { version = "0.35.0", path = "transports/tcp", default-features = false, optional = true }
 libp2p-websocket = { version = "0.37.0", path = "transports/websocket", optional = true }
 
 [target.'cfg(not(target_os = "unknown"))'.dependencies]

README.md

@@ -2,6 +2,8 @@
 
 <a href="http://libp2p.io/"><img src="https://img.shields.io/badge/project-libp2p-yellow.svg?style=flat-square" /></a>
 [![dependency status](https://deps.rs/repo/github/libp2p/rust-libp2p/status.svg?style=flat-square)](https://deps.rs/repo/github/libp2p/rust-libp2p)
+[![Crates.io](https://img.shields.io/crates/v/libp2p.svg)](https://crates.io/crates/libp2p)
+[![docs.rs](https://img.shields.io/badge/api-rustdoc-blue.svg)](https://docs.rs/libp2p)
 
 This repository is the central place for Rust development of the [libp2p](https://libp2p.io) spec.
 

core/CHANGELOG.md

@@ -2,8 +2,13 @@
 
 - Remove `StreamMuxer::poll_event` in favor of individual functions: `poll_inbound`, `poll_outbound`
   and `poll_address_change`. Consequently, `StreamMuxerEvent` is also removed. See [PR 2724].
+- Drop `Unpin` requirement from `SubstreamBox`. See [PR 2762] and [PR 2776].
+- Drop `Sync` requirement on `StreamMuxer` for constructing `StreamMuxerBox`. See [PR 2775].
 
 [PR 2724]: https://github.com/libp2p/rust-libp2p/pull/2724
+[PR 2762]: https://github.com/libp2p/rust-libp2p/pull/2762
+[PR 2775]: https://github.com/libp2p/rust-libp2p/pull/2775
+[PR 2776]: https://github.com/libp2p/rust-libp2p/pull/2776
 
 # 0.34.0
 

core/src/identity/ecdsa.rs

@@ -157,7 +157,7 @@ impl PublicKey {
         let buf = Self::del_asn1_header(k).ok_or_else(|| {
             DecodingError::new("failed to parse asn.1 encoded ecdsa p256 public key")
         })?;
-        Self::from_bytes(&buf)
+        Self::from_bytes(buf)
     }
 
     // ecPublicKey (ANSI X9.62 public key type) OID: 1.2.840.10045.2.1
@@ -198,8 +198,8 @@ impl PublicKey {
         if asn1_head[0] != 0x30
             || asn1_head[2] != 0x30
             || asn1_head[3] as usize != oids_len
-            || &oids_buf[..Self::EC_PUBLIC_KEY_OID.len()] != &Self::EC_PUBLIC_KEY_OID
-            || &oids_buf[Self::EC_PUBLIC_KEY_OID.len()..] != &Self::SECP_256_R1_OID
+            || oids_buf[..Self::EC_PUBLIC_KEY_OID.len()] != Self::EC_PUBLIC_KEY_OID
+            || oids_buf[Self::EC_PUBLIC_KEY_OID.len()..] != Self::SECP_256_R1_OID
             || bitstr_head[0] != 0x03
             || bitstr_head[2] != 0x00
         {

core/src/muxing/boxed.rs

@@ -10,14 +10,14 @@ use std::task::{Context, Poll};
 
 /// Abstract `StreamMuxer`.
 pub struct StreamMuxerBox {
-    inner: Box<dyn StreamMuxer<Substream = SubstreamBox, Error = io::Error> + Send + Sync>,
+    inner: Box<dyn StreamMuxer<Substream = SubstreamBox, Error = io::Error> + Send>,
 }
 
 /// Abstract type for asynchronous reading and writing.
 ///
 /// A [`SubstreamBox`] erases the concrete type it is given and only retains its `AsyncRead`
 /// and `AsyncWrite` capabilities.
-pub struct SubstreamBox(Box<dyn AsyncReadWrite + Send + Unpin>);
+pub struct SubstreamBox(Pin<Box<dyn AsyncReadWrite + Send>>);
 
 struct Wrap<T>
 where
@@ -29,7 +29,7 @@ where
 impl<T> StreamMuxer for Wrap<T>
 where
     T: StreamMuxer,
-    T::Substream: Send + Unpin + 'static,
+    T::Substream: Send + 'static,
     T::Error: Send + Sync + 'static,
 {
     type Substream = SubstreamBox;
@@ -70,8 +70,8 @@ impl StreamMuxerBox {
     /// Turns a stream muxer into a `StreamMuxerBox`.
     pub fn new<T>(muxer: T) -> StreamMuxerBox
     where
-        T: StreamMuxer + Send + Sync + 'static,
-        T::Substream: Send + Unpin + 'static,
+        T: StreamMuxer + Send + 'static,
+        T::Substream: Send + 'static,
         T::Error: Send + Sync + 'static,
     {
         let wrap = Wrap { inner: muxer };
@@ -106,8 +106,8 @@ impl StreamMuxer for StreamMuxerBox {
 
 impl SubstreamBox {
     /// Construct a new [`SubstreamBox`] from something that implements [`AsyncRead`] and [`AsyncWrite`].
-    pub fn new<S: AsyncRead + AsyncWrite + Send + Unpin + 'static>(stream: S) -> Self {
-        Self(Box::new(stream))
+    pub fn new<S: AsyncRead + AsyncWrite + Send + 'static>(stream: S) -> Self {
+        Self(Box::pin(stream))
     }
 }
 
@@ -118,7 +118,7 @@ impl fmt::Debug for SubstreamBox {
 }
 
 /// Workaround because Rust does not allow `Box<dyn AsyncRead + AsyncWrite>`.
-trait AsyncReadWrite: AsyncRead + AsyncWrite + Unpin {
+trait AsyncReadWrite: AsyncRead + AsyncWrite {
     /// Helper function to capture the erased inner type.
     ///
     /// Used to make the [`Debug`] implementation of [`SubstreamBox`] more useful.
@@ -127,7 +127,7 @@ trait AsyncReadWrite: AsyncRead + AsyncWrite + Unpin {
 
 impl<S> AsyncReadWrite for S
 where
-    S: AsyncRead + AsyncWrite + Unpin,
+    S: AsyncRead + AsyncWrite,
 {
     fn type_name(&self) -> &'static str {
         std::any::type_name::<S>()
@@ -136,44 +136,44 @@ where
 
 impl AsyncRead for SubstreamBox {
     fn poll_read(
-        self: Pin<&mut Self>,
+        mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         buf: &mut [u8],
     ) -> Poll<std::io::Result<usize>> {
-        Pin::new(&mut self.get_mut().0).poll_read(cx, buf)
+        self.0.as_mut().poll_read(cx, buf)
     }
 
     fn poll_read_vectored(
-        self: Pin<&mut Self>,
+        mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         bufs: &mut [IoSliceMut<'_>],
     ) -> Poll<std::io::Result<usize>> {
-        Pin::new(&mut self.get_mut().0).poll_read_vectored(cx, bufs)
+        self.0.as_mut().poll_read_vectored(cx, bufs)
     }
 }
 
 impl AsyncWrite for SubstreamBox {
     fn poll_write(
-        self: Pin<&mut Self>,
+        mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<std::io::Result<usize>> {
-        Pin::new(&mut self.get_mut().0).poll_write(cx, buf)
+        self.0.as_mut().poll_write(cx, buf)
     }
 
     fn poll_write_vectored(
-        self: Pin<&mut Self>,
+        mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         bufs: &[IoSlice<'_>],
     ) -> Poll<std::io::Result<usize>> {
-        Pin::new(&mut self.get_mut().0).poll_write_vectored(cx, bufs)
+        self.0.as_mut().poll_write_vectored(cx, bufs)
     }
 
-    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
-        Pin::new(&mut self.get_mut().0).poll_flush(cx)
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        self.0.as_mut().poll_flush(cx)
     }
 
-    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
-        Pin::new(&mut self.get_mut().0).poll_close(cx)
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        self.0.as_mut().poll_close(cx)
     }
 }

core/src/transport/upgrade.rs

@@ -299,8 +299,8 @@ impl<T> Multiplexed<T> {
         T::Dial: Send + 'static,
         T::ListenerUpgrade: Send + 'static,
         T::Error: Send + Sync,
-        M: StreamMuxer + Send + Sync + 'static,
-        M::Substream: Send + Unpin + 'static,
+        M: StreamMuxer + Send + 'static,
+        M::Substream: Send + 'static,
         M::Error: Send + Sync + 'static,
     {
         boxed(self.map(|(i, m), _| (i, StreamMuxerBox::new(m))))

core/tests/transport_upgrade.rs

@@ -18,8 +18,6 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-mod util;
-
 use futures::prelude::*;
 use libp2p_core::identity;
 use libp2p_core::transport::{MemoryTransport, Transport};
@@ -91,11 +89,6 @@ fn upgrade_pipeline() {
         .apply(HelloUpgrade {})
         .apply(HelloUpgrade {})
         .multiplex(MplexConfig::default())
-        .and_then(|(peer, mplex), _| {
-            // Gracefully close the connection to allow protocol
-            // negotiation to complete.
-            util::CloseMuxer::new(mplex).map_ok(move |mplex| (peer, mplex))
-        })
         .boxed();
 
     let dialer_keys = identity::Keypair::generate_ed25519();
@@ -110,11 +103,6 @@ fn upgrade_pipeline() {
         .apply(HelloUpgrade {})
         .apply(HelloUpgrade {})
         .multiplex(MplexConfig::default())
-        .and_then(|(peer, mplex), _| {
-            // Gracefully close the connection to allow protocol
-            // negotiation to complete.
-            util::CloseMuxer::new(mplex).map_ok(move |mplex| (peer, mplex))
-        })
         .boxed();
 
     let listen_addr1 = Multiaddr::from(Protocol::Memory(random::<u64>()));

misc/metrics/CHANGELOG.md

@@ -16,6 +16,10 @@
 
 - Update to `libp2p-core` `v0.35.0`.
 
+- Update to `prometheus-client` `v0.17.0`. See [PR 2761].
+
+[PR 2761]: https://github.com/libp2p/rust-libp2p/pull/2761/
+
 [PR 2734]: https://github.com/libp2p/rust-libp2p/pull/2734/
 
 # 0.7.0

misc/metrics/Cargo.toml

@@ -26,7 +26,7 @@ libp2p-kad = { version = "0.39.0", path = "../../protocols/kad", optional = true
 libp2p-ping = { version = "0.38.0", path = "../../protocols/ping", optional = true }
 libp2p-relay =  { version = "0.11.0", path = "../../protocols/relay", optional = true }
 libp2p-swarm = { version = "0.38.0", path = "../../swarm" }
-prometheus-client = "0.16.0"
+prometheus-client = "0.17.0"
 
 [target.'cfg(not(target_os = "unknown"))'.dependencies]
 libp2p-gossipsub =  { version = "0.40.0", path = "../../protocols/gossipsub", optional = true }

misc/metrics/src/identify.rs

@@ -223,7 +223,7 @@ impl EncodeMetric for Protocols {
                 |mut acc, (_, protocols)| {
                     for protocol in protocols {
                         let count = acc.entry(protocol.to_string()).or_default();
-                        *count = *count + 1;
+                        *count += 1;
                     }
                     acc
                 },

muxers/mplex/tests/async_write.rs

@@ -22,7 +22,6 @@ use futures::future::poll_fn;
 use futures::{channel::oneshot, prelude::*};
 use libp2p_core::{upgrade, StreamMuxer, Transport};
 use libp2p_tcp::TcpTransport;
-use std::sync::Arc;
 
 #[test]
 fn async_write() {
@@ -72,7 +71,7 @@ fn async_write() {
         let mut transport = TcpTransport::default()
             .and_then(move |c, e| upgrade::apply(c, mplex, e, upgrade::Version::V1));
 
-        let client = Arc::new(transport.dial(rx.await.unwrap()).unwrap().await.unwrap());
+        let client = transport.dial(rx.await.unwrap()).unwrap().await.unwrap();
         let mut inbound = poll_fn(|cx| client.poll_inbound(cx)).await.unwrap();
         inbound.write_all(b"hello world").await.unwrap();
 

muxers/mplex/tests/two_peers.rs

@@ -22,7 +22,6 @@ use futures::future::poll_fn;
 use futures::{channel::oneshot, prelude::*};
 use libp2p_core::{upgrade, StreamMuxer, Transport};
 use libp2p_tcp::TcpTransport;
-use std::sync::Arc;
 
 #[test]
 fn client_to_server_outbound() {
@@ -73,7 +72,7 @@ fn client_to_server_outbound() {
             .and_then(move |c, e| upgrade::apply(c, mplex, e, upgrade::Version::V1))
             .boxed();
 
-        let client = Arc::new(transport.dial(rx.await.unwrap()).unwrap().await.unwrap());
+        let client = transport.dial(rx.await.unwrap()).unwrap().await.unwrap();
         let mut inbound = poll_fn(|cx| client.poll_inbound(cx)).await.unwrap();
         inbound.write_all(b"hello world").await.unwrap();
         inbound.close().await.unwrap();
@@ -108,17 +107,15 @@ fn client_to_server_inbound() {
 
         tx.send(addr).unwrap();
 
-        let client = Arc::new(
-            transport
-                .next()
-                .await
-                .expect("some event")
-                .into_incoming()
-                .unwrap()
-                .0
-                .await
-                .unwrap(),
-        );
+        let client = transport
+            .next()
+            .await
+            .expect("some event")
+            .into_incoming()
+            .unwrap()
+            .0
+            .await
+            .unwrap();
 
         let mut inbound = poll_fn(|cx| client.poll_inbound(cx)).await.unwrap();
 

protocols/gossipsub/CHANGELOG.md

@@ -4,6 +4,10 @@
 
 - Update to `libp2p-core` `v0.35.0`.
 
+- Update to `prometheus-client` `v0.17.0`. See [PR 2761].
+
+[PR 2761]: https://github.com/libp2p/rust-libp2p/pull/2761/
+
 # 0.39.0
 
 - Update to `libp2p-core` `v0.34.0`.

protocols/gossipsub/Cargo.toml

@@ -31,7 +31,7 @@ serde = { version = "1", optional = true, features = ["derive"] }
 wasm-timer = "0.2.5"
 instant = "0.1.11"
 # Metrics dependencies
-prometheus-client = "0.16.0"
+prometheus-client = "0.17.0"
 
 [dev-dependencies]
 async-std = "1.6.3"

protocols/mdns/CHANGELOG.md

@@ -1,6 +1,7 @@
 # 0.39.0 [unreleased]
 
 - Update to `libp2p-swarm` `v0.38.0`.
+- Update to `if-watch` `v1.1.1`.
 
 - Allow users to choose between async-io and tokio runtime
   in the mdns protocol implementation. `async-io` is a default

protocols/mdns/Cargo.toml

@@ -14,7 +14,7 @@ categories = ["network-programming", "asynchronous"]
 data-encoding = "2.3.2"
 dns-parser = "0.8.0"
 futures = "0.3.13"
-if-watch = "1.0.0"
+if-watch = "1.1.1"
 lazy_static = "1.4.0"
 libp2p-core = { version = "0.35.0", path = "../../core", default-features = false }
 libp2p-swarm = { version = "0.38.0", path = "../../swarm" }