Update to bytes v0.5

Except for `multiaddr` which encapsulates its use of bytes v0.4 now.

Cargo.toml

@@ -14,7 +14,7 @@ default = ["secp256k1", "libp2p-websocket"]
 secp256k1 = ["libp2p-core/secp256k1", "libp2p-secio/secp256k1"]
 
 [dependencies]
-bytes = "0.4"
+bytes = "0.5"
 futures = "0.3.1"
 multiaddr = { package = "parity-multiaddr", version = "0.6.0", path = "misc/multiaddr" }
 multihash = { package = "parity-multihash", version = "0.2.0", path = "misc/multihash" }

core/Cargo.toml

@@ -12,7 +12,7 @@ categories = ["network-programming", "asynchronous"]
 [dependencies]
 asn1_der = "0.6.1"
 bs58 = "0.3.0"
-bytes = "0.4"
+bytes = "0.5"
 ed25519-dalek = "1.0.0-pre.3"
 failure = "0.1"
 fnv = "1.0"
@@ -32,7 +32,7 @@ rand = "0.7"
 rw-stream-sink = { version = "0.1.1", path = "../misc/rw-stream-sink" }
 sha2 = "0.8.0"
 smallvec = "1.0"
-unsigned-varint = "0.2"
+unsigned-varint = { git = "https://github.com/twittner/unsigned-varint.git", branch = "bytes-0.5" }
 void = "1"
 zeroize = "1"
 

core/src/transport/memory.rs

@@ -19,7 +19,6 @@
 // DEALINGS IN THE SOFTWARE.
 
 use crate::{Transport, transport::{TransportError, ListenerEvent}};
-use bytes::IntoBuf;
 use fnv::FnvHashMap;
 use futures::{future::{self, Ready}, prelude::*, channel::mpsc, task::Context, task::Poll};
 use lazy_static::lazy_static;
@@ -271,8 +270,7 @@ impl<T> Sink<T> for Chan<T> {
     }
 }
 
-impl<T: IntoBuf> Into<RwStreamSink<Chan<T>>> for Chan<T> {
+impl<T: AsRef<[u8]>> Into<RwStreamSink<Chan<T>>> for Chan<T> {
-    #[inline]
     fn into(self) -> RwStreamSink<Chan<T>> {
         RwStreamSink::new(self)
     }

misc/multiaddr/src/lib.rs

@@ -7,7 +7,7 @@ mod errors;
 mod from_url;
 mod util;
 
-use bytes::{Bytes, BytesMut};
+use bytes::Bytes;
 use serde::{
     Deserialize,
     Deserializer,
@@ -290,10 +290,10 @@ impl From<Ipv6Addr> for Multiaddr {
     }
 }
 
-impl TryFrom<Bytes> for Multiaddr {
+impl TryFrom<Vec<u8>> for Multiaddr {
     type Error = Error;
 
-    fn try_from(v: Bytes) -> Result<Self> {
+    fn try_from(v: Vec<u8>) -> Result<Self> {
         // Check if the argument is a valid `Multiaddr` by reading its protocols.
         let mut slice = &v[..];
         while !slice.is_empty() {
@@ -304,22 +304,6 @@ impl TryFrom<Bytes> for Multiaddr {
     }
 }
 
-impl TryFrom<BytesMut> for Multiaddr {
-    type Error = Error;
-
-    fn try_from(v: BytesMut) -> Result<Self> {
-        Multiaddr::try_from(v.freeze())
-    }
-}
-
-impl TryFrom<Vec<u8>> for Multiaddr {
-    type Error = Error;
-
-    fn try_from(v: Vec<u8>) -> Result<Self> {
-        Multiaddr::try_from(Bytes::from(v))
-    }
-}
-
 impl TryFrom<String> for Multiaddr {
     type Error = Error;
 

misc/multihash/Cargo.toml

@@ -11,9 +11,9 @@ documentation = "https://docs.rs/parity-multihash/"
 
 [dependencies]
 blake2 = { version = "0.8", default-features = false }
-bytes = "0.4.12"
+bytes = "0.5"
-rand = { version = "0.6", default-features = false, features = ["std"] }
+rand = { version = "0.7", default-features = false, features = ["std"] }
 sha-1 = { version = "0.8", default-features = false }
 sha2 = { version = "0.8", default-features = false }
 sha3 = { version = "0.8", default-features = false }
-unsigned-varint = "0.2"
+unsigned-varint = { git = "https://github.com/twittner/unsigned-varint.git", branch = "bytes-0.5" }

misc/multihash/src/lib.rs

@@ -247,7 +247,7 @@ impl<'a> MultihashRef<'a> {
     /// This operation allocates.
     pub fn into_owned(self) -> Multihash {
         Multihash {
-            bytes: Bytes::from(self.bytes)
+            bytes: Bytes::copy_from_slice(self.bytes)
         }
     }
 

misc/multistream-select/Cargo.toml

@@ -10,12 +10,12 @@ categories = ["network-programming", "asynchronous"]
 edition = "2018"
 
 [dependencies]
-bytes = "0.4"
+bytes = "0.5"
 futures = "0.1"
 log = "0.4"
 smallvec = "1.0"
 tokio-io = "0.1"
-unsigned-varint = "0.2.2"
+unsigned-varint = { git = "https://github.com/twittner/unsigned-varint.git", branch = "bytes-0.5" }
 
 [dev-dependencies]
 tokio = "0.1"

misc/multistream-select/src/length_delimited.rs

@@ -18,7 +18,7 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use bytes::{Bytes, BytesMut, BufMut};
+use bytes::{Bytes, BytesMut, Buf, BufMut};
 use futures::{try_ready, Async, Poll, Sink, StartSend, Stream, AsyncSink};
 use std::{io, u16};
 use tokio_io::{AsyncRead, AsyncWrite};
@@ -136,7 +136,7 @@ impl<R> LengthDelimited<R> {
                     "Failed to write buffered frame."))
             }
 
-            self.write_buffer.split_to(n);
+            self.write_buffer.advance(n);
         }
 
         Ok(Async::Ready(()))

misc/multistream-select/src/negotiated.rs

@@ -18,7 +18,7 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use bytes::BytesMut;
+use bytes::{BytesMut, Buf};
 use crate::protocol::{Protocol, MessageReader, Message, Version, ProtocolError};
 use futures::{prelude::*, Async, try_ready};
 use log::debug;
@@ -93,7 +93,7 @@ impl<TInner> Negotiated<TInner> {
         }
 
         if let State::Completed { remaining, .. } = &mut self.state {
-            let _ = remaining.take(); // Drop remaining data flushed above.
+            let _ = remaining.split_to(remaining.len()); // Drop remaining data flushed above.
             return Ok(Async::Ready(()))
         }
 
@@ -232,7 +232,7 @@ where
                     if n == 0 {
                         return Err(io::ErrorKind::WriteZero.into())
                     }
-                    remaining.split_to(n);
+                    remaining.advance(n);
                 }
                 io.write(buf)
             },
@@ -251,7 +251,7 @@ where
                             io::ErrorKind::WriteZero,
                             "Failed to write remaining buffer."))
                     }
-                    remaining.split_to(n);
+                    remaining.advance(n);
                 }
                 io.flush()
             },
@@ -363,7 +363,7 @@ mod tests {
             let cap = rem.len() + free as usize;
             let step = u8::min(free, step) as usize + 1;
             let buf = Capped { buf: Vec::with_capacity(cap), step };
-            let rem = BytesMut::from(rem);
+            let rem = BytesMut::from(&rem[..]);
             let mut io = Negotiated::completed(buf, rem.clone());
             let mut written = 0;
             loop {

misc/multistream-select/src/protocol.rs

@@ -143,7 +143,7 @@ impl TryFrom<&[u8]> for Protocol {
     type Error = ProtocolError;
 
     fn try_from(value: &[u8]) -> Result<Self, Self::Error> {
-        Self::try_from(Bytes::from(value))
+        Self::try_from(Bytes::copy_from_slice(value))
     }
 }
 
@@ -208,7 +208,7 @@ impl Message {
                     out_msg.push(b'\n')
                 }
                 dest.reserve(out_msg.len());
-                dest.put(out_msg);
+                dest.put(out_msg.as_ref());
                 Ok(())
             }
             Message::NotAvailable => {
@@ -254,7 +254,7 @@ impl Message {
             if len == 0 || len > rem.len() || rem[len - 1] != b'\n' {
                 return Err(ProtocolError::InvalidMessage)
             }
-            let p = Protocol::try_from(Bytes::from(&rem[.. len - 1]))?;
+            let p = Protocol::try_from(Bytes::copy_from_slice(&rem[.. len - 1]))?;
             protocols.push(p);
             remaining = &rem[len ..]
         }

misc/rw-stream-sink/Cargo.toml

@@ -10,8 +10,8 @@ keywords = ["networking"]
 categories = ["network-programming", "asynchronous"]
 
 [dependencies]
-bytes = "0.4.12"
 futures = "0.3.1"
+static_assertions = "1"
 
 [dev-dependencies]
 async-std = "1.0"

misc/rw-stream-sink/src/lib.rs

@@ -25,26 +25,19 @@
 //! Each call to [`AsyncWrite::poll_write`] will send one packet to the sink.
 //! Calls to [`AsyncRead::read`] will read from the stream's incoming packets.
 
-use bytes::{IntoBuf, Buf};
 use futures::{prelude::*, ready};
-use std::{io, pin::Pin, task::{Context, Poll}};
+use std::{io::{self, Read}, pin::Pin, task::{Context, Poll}};
+
+static_assertions::const_assert!(std::mem::size_of::<usize>() <= std::mem::size_of::<u64>());
 
 /// Wraps a [`Stream`] and [`Sink`] whose items are buffers.
 /// Implements [`AsyncRead`] and [`AsyncWrite`].
-pub struct RwStreamSink<S>
+pub struct RwStreamSink<S: TryStream> {
-where
-    S: TryStream,
-    <S as TryStream>::Ok: IntoBuf
-{
     inner: S,
-    current_item: Option<<<S as TryStream>::Ok as IntoBuf>::Buf>
+    current_item: Option<std::io::Cursor<<S as TryStream>::Ok>>
 }
 
-impl<S> RwStreamSink<S>
+impl<S: TryStream> RwStreamSink<S> {
-where
-    S: TryStream,
-    <S as TryStream>::Ok: IntoBuf
-{
     /// Wraps around `inner`.
     pub fn new(inner: S) -> Self {
         RwStreamSink { inner, current_item: None }
@@ -54,35 +47,32 @@ where
 impl<S> AsyncRead for RwStreamSink<S>
 where
     S: TryStream<Error = io::Error> + Unpin,
-    <S as TryStream>::Ok: IntoBuf
+    <S as TryStream>::Ok: AsRef<[u8]>
 {
     fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context, buf: &mut [u8]) -> Poll<io::Result<usize>> {
         // Grab the item to copy from.
         let item_to_copy = loop {
             if let Some(ref mut i) = self.current_item {
-                if i.has_remaining() {
+                if i.position() < i.get_ref().as_ref().len() as u64 {
                     break i
                 }
             }
             self.current_item = Some(match ready!(self.inner.try_poll_next_unpin(cx)) {
-                Some(Ok(i)) => i.into_buf(),
+                Some(Ok(i)) => std::io::Cursor::new(i),
                 Some(Err(e)) => return Poll::Ready(Err(e)),
                 None => return Poll::Ready(Ok(0)) // EOF
             });
         };
 
         // Copy it!
-        debug_assert!(item_to_copy.has_remaining());
+        Poll::Ready(Ok(item_to_copy.read(buf)?))
-        let to_copy = std::cmp::min(buf.len(), item_to_copy.remaining());
-        item_to_copy.take(to_copy).copy_to_slice(&mut buf[.. to_copy]);
-        Poll::Ready(Ok(to_copy))
     }
 }
 
 impl<S> AsyncWrite for RwStreamSink<S>
 where
     S: TryStream + Sink<<S as TryStream>::Ok, Error = io::Error> + Unpin,
-    <S as TryStream>::Ok: IntoBuf + for<'r> From<&'r [u8]>
+    <S as TryStream>::Ok: for<'r> From<&'r [u8]>
 {
     fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context, buf: &[u8]) -> Poll<io::Result<usize>> {
         ready!(Pin::new(&mut self.inner).poll_ready(cx)?);
@@ -102,16 +92,11 @@ where
     }
 }
 
-impl<S> Unpin for RwStreamSink<S>
+impl<S: TryStream> Unpin for RwStreamSink<S> {}
-where
-    S: TryStream,
-    <S as TryStream>::Ok: IntoBuf
-{}
 
 #[cfg(test)]
 mod tests {
     use async_std::task;
-    use bytes::Bytes;
     use futures::{channel::mpsc, prelude::*, stream};
     use std::{pin::Pin, task::{Context, Poll}};
     use super::RwStreamSink;
@@ -163,9 +148,9 @@ mod tests {
         let mut wrapper = RwStreamSink::new(Wrapper(rx2.map(Ok), tx1));
 
         task::block_on(async move {
-            tx2.send(Bytes::from("hel")).await.unwrap();
+            tx2.send(Vec::from("hel")).await.unwrap();
-            tx2.send(Bytes::from("lo wor")).await.unwrap();
+            tx2.send(Vec::from("lo wor")).await.unwrap();
-            tx2.send(Bytes::from("ld")).await.unwrap();
+            tx2.send(Vec::from("ld")).await.unwrap();
             tx2.close().await.unwrap();
 
             let mut data = Vec::new();

muxers/mplex/Cargo.toml

@@ -10,14 +10,14 @@ keywords = ["peer-to-peer", "libp2p", "networking"]
 categories = ["network-programming", "asynchronous"]
 
 [dependencies]
-bytes = "0.4.5"
+bytes = "0.5"
 fnv = "1.0"
 futures = "0.3.1"
-futures_codec = "= 0.3.3"
+futures_codec = "0.3.4"
 libp2p-core = { version = "0.13.0", path = "../../core" }
 log = "0.4"
 parking_lot = "0.9"
-unsigned-varint = { version = "0.2.3", features = ["futures-codec"] }
+unsigned-varint = { git = "https://github.com/twittner/unsigned-varint.git", branch = "bytes-0.5", features = ["futures-codec"] }
 
 [dev-dependencies]
 async-std = "1.0"

muxers/mplex/src/lib.rs

@@ -535,7 +535,7 @@ where C: AsyncRead + AsyncWrite + Unpin
 
         let elem = codec::Elem::Data {
             substream_id: substream.num,
-            data: From::from(&buf[..to_write]),
+            data: Bytes::copy_from_slice(&buf[..to_write]),
             endpoint: substream.endpoint,
         };
 

protocols/floodsub/Cargo.toml

@@ -11,12 +11,12 @@ categories = ["network-programming", "asynchronous"]
 
 [dependencies]
 bs58 = "0.3.0"
-bytes = "0.4"
+bytes = "0.5"
 cuckoofilter = "0.3.2"
 fnv = "1.0"
 futures = "0.3.1"
 libp2p-core = { version = "0.13.0", path = "../../core" }
 libp2p-swarm = { version = "0.3.0", path = "../../swarm" }
 protobuf = "2.8"
-rand = "0.6"
+rand = "0.7"
 smallvec = "1.0"

protocols/identify/Cargo.toml

@@ -10,8 +10,8 @@ keywords = ["peer-to-peer", "libp2p", "networking"]
 categories = ["network-programming", "asynchronous"]
 
 [dependencies]
-bytes = "0.4"
+bytes = "0.5"
-futures_codec = "= 0.3.3"
+futures_codec = "0.3.4"
 futures = "0.3.1"
 libp2p-core = { version = "0.13.0", path = "../../core" }
 libp2p-swarm = { version = "0.3.0", path = "../../swarm" }
@@ -20,7 +20,7 @@ multiaddr = { package = "parity-multiaddr", version = "0.6.0", path = "../../mis
 protobuf = "2.8"
 smallvec = "1.0"
 wasm-timer = "0.2"
-unsigned-varint = { version = "0.2.3", features = ["futures-codec"] }
+unsigned-varint = { git = "https://github.com/twittner/unsigned-varint.git", branch = "bytes-0.5", features = ["futures-codec"] }
 
 [dev-dependencies]
 async-std = "1.0"

protocols/kad/Cargo.toml

@@ -11,10 +11,10 @@ categories = ["network-programming", "asynchronous"]
 
 [dependencies]
 arrayvec = "0.5.1"
-bytes = "0.4"
+bytes = "0.5"
 either = "1.5"
 fnv = "1.0"
-futures_codec = "= 0.3.3"
+futures_codec = "0.3.4"
 futures = "0.3.1"
 log = "0.4"
 libp2p-core = { version = "0.13.0", path = "../../core" }
@@ -27,7 +27,7 @@ sha2 = "0.8.0"
 smallvec = "1.0"
 wasm-timer = "0.2"
 uint = "0.8"
-unsigned-varint = { version = "0.2.3", features = ["futures-codec"] }
+unsigned-varint = { git = "https://github.com/twittner/unsigned-varint.git", branch = "bytes-0.5", features = ["futures-codec"] }
 void = "1.0"
 
 [dev-dependencies]

protocols/kad/src/protocol.rs

@@ -58,7 +58,6 @@ pub enum KadConnectionType {
 }
 
 impl From<proto::Message_ConnectionType> for KadConnectionType {
-    #[inline]
     fn from(raw: proto::Message_ConnectionType) -> KadConnectionType {
         use proto::Message_ConnectionType::{
             CAN_CONNECT, CANNOT_CONNECT, CONNECTED, NOT_CONNECTED
@@ -73,7 +72,6 @@ impl From<proto::Message_ConnectionType> for KadConnectionType {
 }
 
 impl Into<proto::Message_ConnectionType> for KadConnectionType {
-    #[inline]
     fn into(self) -> proto::Message_ConnectionType {
         use proto::Message_ConnectionType::{
             CAN_CONNECT, CANNOT_CONNECT, CONNECTED, NOT_CONNECTED
@@ -181,7 +179,6 @@ where
     type Future = future::Ready<Result<Self::Output, io::Error>>;
     type Error = io::Error;
 
-    #[inline]
     fn upgrade_inbound(self, incoming: Negotiated<C>, _: Self::Info) -> Self::Future {
         let mut codec = UviBytes::default();
         codec.set_max_len(4096);
@@ -191,7 +188,9 @@ where
                 .err_into()
                 .with::<_, _, fn(_) -> _, _>(|response| {
                     let proto_struct = resp_msg_to_proto(response);
-                    future::ready(proto_struct.write_to_bytes().map_err(invalid_data))
+                    future::ready(proto_struct.write_to_bytes()
+                        .map(io::Cursor::new)
+                        .map_err(invalid_data))
                 })
                 .and_then::<_, fn(_) -> _>(|bytes| {
                     let request = match protobuf::parse_from_bytes(&bytes) {
@@ -212,7 +211,6 @@ where
     type Future = future::Ready<Result<Self::Output, io::Error>>;
     type Error = io::Error;
 
-    #[inline]
     fn upgrade_outbound(self, incoming: Negotiated<C>, _: Self::Info) -> Self::Future {
         let mut codec = UviBytes::default();
         codec.set_max_len(4096);
@@ -222,7 +220,9 @@ where
                 .err_into()
                 .with::<_, _, fn(_) -> _, _>(|request| {
                     let proto_struct = req_msg_to_proto(request);
-                    future::ready(proto_struct.write_to_bytes().map_err(invalid_data))
+                    future::ready(proto_struct.write_to_bytes()
+                        .map(io::Cursor::new)
+                        .map_err(invalid_data))
                 })
                 .and_then::<_, fn(_) -> _>(|bytes| {
                     let response = match protobuf::parse_from_bytes(&bytes) {
@@ -243,11 +243,11 @@ pub type KadOutStreamSink<S> = KadStreamSink<S, KadRequestMsg, KadResponseMsg>;
 
 pub type KadStreamSink<S, A, B> = stream::AndThen<
     sink::With<
-        stream::ErrInto<Framed<S, UviBytes<Vec<u8>>>, io::Error>,
+        stream::ErrInto<Framed<S, UviBytes<io::Cursor<Vec<u8>>>>, io::Error>,
-        Vec<u8>,
+        io::Cursor<Vec<u8>>,
         A,
-        future::Ready<Result<Vec<u8>, io::Error>>,
+        future::Ready<Result<io::Cursor<Vec<u8>>, io::Error>>,
-        fn(A) -> future::Ready<Result<Vec<u8>, io::Error>>,
+        fn(A) -> future::Ready<Result<io::Cursor<Vec<u8>>, io::Error>>,
     >,
     future::Ready<Result<B, io::Error>>,
     fn(BytesMut) -> future::Ready<Result<B, io::Error>>,

protocols/kad/src/record.rs

@@ -35,7 +35,7 @@ pub struct Key(Bytes);
 impl Key {
     /// Creates a new key from the bytes of the input.
     pub fn new<K: AsRef<[u8]>>(key: &K) -> Self {
-        Key(Bytes::from(key.as_ref()))
+        Key(Bytes::copy_from_slice(key.as_ref()))
     }
 
     /// Copies the bytes of the key into a new vector.

protocols/noise/Cargo.toml

@@ -8,7 +8,7 @@ repository = "https://github.com/libp2p/rust-libp2p"
 edition = "2018"
 
 [dependencies]
-bytes = "0.4"
+bytes = "0.5"
 curve25519-dalek = "1"
 futures = "0.3.1"
 lazy_static = "1.2"

protocols/ping/Cargo.toml

@@ -10,15 +10,15 @@ keywords = ["peer-to-peer", "libp2p", "networking"]
 categories = ["network-programming", "asynchronous"]
 
 [dependencies]
-bytes = "0.4"
+bytes = "0.5"
+futures = "0.3.1"
 libp2p-core = { version = "0.13.0", path = "../../core" }
 libp2p-swarm = { version = "0.3.0", path = "../../swarm" }
 log = "0.4.1"
 multiaddr = { package = "parity-multiaddr", version = "0.6.0", path = "../../misc/multiaddr" }
-futures = "0.3.1"
 rand = "0.7.2"
-wasm-timer = "0.2"
 void = "1.0"
+wasm-timer = "0.2"
 
 [dev-dependencies]
 async-std = "1.0"

protocols/plaintext/Cargo.toml

@@ -10,14 +10,14 @@ keywords = ["peer-to-peer", "libp2p", "networking"]
 categories = ["network-programming", "asynchronous"]
 
 [dependencies]
-bytes = "0.4.12"
+bytes = "0.5"
 futures = "0.3.1"
-futures_codec = "= 0.3.3"
+futures_codec = "0.3.4"
 libp2p-core = { version = "0.13.0", path = "../../core" }
 log = "0.4.8"
 protobuf = "2.8.1"
 rw-stream-sink = { version = "0.1.1", path = "../../misc/rw-stream-sink" }
-unsigned-varint = { version = "0.2.3", features = ["futures-codec"] }
+unsigned-varint = { git = "https://github.com/twittner/unsigned-varint.git", branch = "bytes-0.5", features = ["futures-codec"] }
 void = "1.0.2"
 
 [dev-dependencies]

protocols/plaintext/src/handshake.rs

@@ -120,7 +120,7 @@ where
     let context = HandshakeContext::new(config)?;
 
     trace!("sending exchange to remote");
-    socket.send(BytesMut::from(context.state.exchange_bytes.clone())).await?;
+    socket.send(BytesMut::from(&context.state.exchange_bytes[..])).await?;
 
     trace!("receiving the remote's exchange");
     let context = match socket.next().await {

transports/tcp/Cargo.toml

@@ -11,7 +11,7 @@ categories = ["network-programming", "asynchronous"]
 
 [dependencies]
 async-std = "1.0"
-bytes = "0.4.12"
+bytes = "0.5"
 futures = "0.3.1"
 futures-timer = "2.0"
 get_if_addrs = "0.5.3"

transports/websocket/Cargo.toml

@@ -11,7 +11,7 @@ categories = ["network-programming", "asynchronous"]
 
 [dependencies]
 async-tls = "0.6"
-bytes = "0.4.12"
+bytes = "0.5"
 either = "1.5.3"
 futures = "0.3.1"
 libp2p-core = { version = "0.13.0", path = "../../core" }