Restore RequestResponse::throttled.

protocols/request-response/Cargo.toml

@@ -11,13 +11,15 @@ categories = ["network-programming", "asynchronous"]
 
 [dependencies]
 async-trait = "0.1"
+bytes = "0.5.6"
 futures = "0.3.1"
 libp2p-core = { version = "0.21.0", path = "../../core" }
 libp2p-swarm = { version = "0.22.0", path = "../../swarm" }
 log = "0.4.11"
-lru = "0.6"
+minicbor = { version = "0.5", features = ["std", "derive"] }
 rand = "0.7"
 smallvec = "1.4"
+unsigned-varint = { version = "0.5", features = ["std", "futures"] }
 wasm-timer = "0.2"
 
 [dev-dependencies]

protocols/request-response/src/codec.rs

@@ -18,6 +18,8 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
+pub(crate) mod header;
+
 pub use libp2p_core::ProtocolName;
 
 use async_trait::async_trait;
@@ -64,3 +66,4 @@ pub trait RequestResponseCodec {
     where
         T: AsyncWrite + Unpin + Send;
 }
+

protocols/request-response/src/codec/header.rs

@@ -0,0 +1,244 @@
+// Copyright 2020 Parity Technologies (UK) Ltd.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+use async_trait::async_trait;
+use bytes::{Bytes, BytesMut};
+use futures::prelude::*;
+use libp2p_core::ProtocolName;
+use minicbor::{Encode, Decode};
+use std::io;
+use super::RequestResponseCodec;
+use unsigned_varint::{aio, io::ReadError};
+
+/// A protocol header.
+#[derive(Debug, Default, Clone, PartialEq, Eq, Encode, Decode)]
+#[cbor(map)]
+pub struct Header {
+    /// The type of message.
+    #[n(0)] pub typ: Option<Type>,
+    /// The number of additional requests the remote is willing to receive.
+    #[n(1)] pub credit: Option<u16>,
+    /// An identifier used for sending credit grants.
+    #[n(2)] pub ident: Option<u64>
+}
+
+/// A protocol message type.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub enum Type {
+    #[n(0)] Request,
+    #[n(1)] Response,
+    #[n(2)] Credit,
+    #[n(3)] Ack
+}
+
+/// A protocol message consisting of header and data.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Message<T> {
+    header: Header,
+    data: Option<T>
+}
+
+impl<T> Message<T> {
+    /// Create a new message of some type.
+    fn new(header: Header) -> Self {
+        Message { header, data: None }
+    }
+
+    /// Create a request message.
+    pub fn request(data: T) -> Self {
+        let mut m = Message::new(Header { typ: Some(Type::Request), .. Header::default() });
+        m.data = Some(data);
+        m
+    }
+
+    /// Create a response message.
+    pub fn response(data: T) -> Self {
+        let mut m = Message::new(Header { typ: Some(Type::Response), .. Header::default() });
+        m.data = Some(data);
+        m
+    }
+
+    /// Create a credit grant.
+    pub fn credit(credit: u16, ident: u64) -> Self {
+        Message::new(Header { typ: Some(Type::Credit), credit: Some(credit), ident: Some(ident) })
+    }
+
+    /// Create an acknowledge message.
+    pub fn ack(ident: u64) -> Self {
+        Message::new(Header { typ: Some(Type::Ack), credit: None, ident: Some(ident) })
+    }
+
+    /// Access the message header.
+    pub fn header(&self) -> &Header {
+        &self.header
+    }
+
+    /// Access the message data.
+    pub fn data(&self) -> Option<&T> {
+        self.data.as_ref()
+    }
+
+    /// Consume this message and return header and data.
+    pub fn into_parts(self) -> (Header, Option<T>) {
+        (self.header, self.data)
+    }
+}
+
+/// A wrapper around a `ProtocolName` impl which augments the protocol name.
+///
+/// The type implements `ProtocolName` itself and creates a name for a
+/// request-response protocol based on the protocol name of the wrapped type.
+#[derive(Debug, Clone)]
+pub struct ProtocolWrapper<P>(P, Bytes);
+
+impl<P: ProtocolName> ProtocolWrapper<P> {
+    pub fn v1(p: P) -> Self {
+        let mut full = BytesMut::from(b"/request-response/header/1".as_ref());
+        full.extend_from_slice(p.protocol_name());
+        ProtocolWrapper(p, full.freeze())
+    }
+}
+
+impl<P> ProtocolName for ProtocolWrapper<P> {
+    fn protocol_name(&self) -> &[u8] {
+        self.1.as_ref()
+    }
+}
+
+/// A `RequestResponseCodec` wrapper that adds headers to the payload data.
+#[derive(Debug, Clone)]
+pub struct Codec<C> {
+    /// The wrapped codec.
+    inner: C,
+    /// Encoding/decoding buffer.
+    buffer: Vec<u8>
+}
+
+impl<C> Codec<C> {
+    /// Create a codec by wrapping an existing one.
+    pub fn new(c: C) -> Self {
+        Codec { inner: c, buffer: Vec::new() }
+    }
+
+    /// Read and decode a request header.
+    async fn read_header<T, H>(&mut self, io: &mut T) -> io::Result<H>
+    where
+        T: AsyncRead + Unpin + Send,
+        H: for<'a> minicbor::Decode<'a>
+    {
+        let header_len = aio::read_u32(&mut *io).await
+            .map_err(|e| match e {
+                ReadError::Io(e) => e,
+                other => io::Error::new(io::ErrorKind::Other, other)
+            })?;
+        self.buffer.resize(u32_to_usize(header_len), 0u8);
+        io.read_exact(&mut self.buffer).await?;
+        minicbor::decode(&self.buffer).map_err(|e| io::Error::new(io::ErrorKind::Other, e))
+    }
+
+    /// Encode and write a response header.
+    async fn write_header<T, H>(&mut self, hdr: &H, io: &mut T) -> io::Result<()>
+    where
+        T: AsyncWrite + Unpin + Send,
+        H: minicbor::Encode
+    {
+        self.buffer.clear();
+        minicbor::encode(hdr, &mut self.buffer).map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
+        let mut b = unsigned_varint::encode::u32_buffer();
+        assert!(self.buffer.len() < u32_to_usize(u32::MAX));
+        let header_len = unsigned_varint::encode::u32(self.buffer.len() as u32, &mut b);
+        io.write_all(header_len).await?;
+        io.write_all(&self.buffer).await
+    }
+}
+
+#[async_trait]
+impl<C> RequestResponseCodec for Codec<C>
+where
+    C: RequestResponseCodec + Send,
+    C::Protocol: Sync
+{
+    type Protocol = ProtocolWrapper<C::Protocol>;
+    type Request = Message<C::Request>;
+    type Response = Message<C::Response>;
+
+    async fn read_request<T>(&mut self, p: &Self::Protocol, io: &mut T) -> io::Result<Self::Request>
+    where
+        T: AsyncRead + Unpin + Send
+    {
+        let mut msg = Message::new(self.read_header(io).await?);
+        match msg.header.typ {
+            Some(Type::Request) => {
+                msg.data = Some(self.inner.read_request(&p.0, io).await?);
+                Ok(msg)
+            }
+            Some(Type::Credit) => Ok(msg),
+            Some(Type::Response) | Some(Type::Ack) | None => {
+                log::debug!("unexpected {:?} when expecting request or credit grant", msg.header.typ);
+                Err(io::ErrorKind::InvalidData.into())
+            }
+        }
+    }
+
+    async fn read_response<T>(&mut self, p: &Self::Protocol, io: &mut T) -> io::Result<Self::Response>
+    where
+        T: AsyncRead + Unpin + Send
+    {
+        let mut msg = Message::new(self.read_header(io).await?);
+        match msg.header.typ {
+            Some(Type::Response) => {
+                msg.data = Some(self.inner.read_response(&p.0, io).await?);
+                Ok(msg)
+            }
+            Some(Type::Ack) => Ok(msg),
+            Some(Type::Request) | Some(Type::Credit) | None => {
+                log::debug!("unexpected {:?} when expecting response or ack", msg.header.typ);
+                Err(io::ErrorKind::InvalidData.into())
+            }
+        }
+    }
+
+    async fn write_request<T>(&mut self, p: &Self::Protocol, io: &mut T, r: Self::Request) -> io::Result<()>
+    where
+        T: AsyncWrite + Unpin + Send
+    {
+        self.write_header(&r.header, io).await?;
+        if let Some(data) = r.data {
+            self.inner.write_request(&p.0, io, data).await?
+        }
+        Ok(())
+    }
+
+    async fn write_response<T>(&mut self, p: &Self::Protocol, io: &mut T, r: Self::Response) -> io::Result<()>
+    where
+        T: AsyncWrite + Unpin + Send
+    {
+        self.write_header(&r.header, io).await?;
+        if let Some(data) = r.data {
+            self.inner.write_response(&p.0, io, data).await?
+        }
+        Ok(())
+    }
+}
+
+#[cfg(any(target_pointer_width = "64", target_pointer_width = "32"))]
+fn u32_to_usize(n: u32) -> usize {
+    n as usize
+}