Make PeerConnector tower::Buffer'able

zebra-network/src/peer/connector.rs

@@ -68,7 +68,8 @@ where
 {
     type Response = PeerClient;
     type Error = HandshakeError;
-    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;
+    type Future =
+        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
 
     fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         Poll::Ready(Ok(()))

zebrad/src/commands/connect.rs

@@ -80,8 +80,11 @@ impl ConnectCmd {
         let mut pc = PeerConnector::new(config, Network::Mainnet, node, &collector);
 
         let tcp_stream = TcpStream::connect(self.addr).await?;
-        pc.ready().await?;
+        pc.ready()
-        let mut client = pc.call((tcp_stream, self.addr)).await?;
+            .await?;
+        let mut client = pc
+            .call((tcp_stream, self.addr))
+            .await?;
 
         client.ready().await?;
 

zebrad/src/commands/listen.rs

@@ -71,18 +71,27 @@ impl ListenCmd {
         );
 
         use tokio::net::{TcpListener, TcpStream};
+        use tokio::prelude::*;
 
         let config = app_config().network.clone();
         let collector = TimestampCollector::new();
-        let mut pc = PeerConnector::new(config, Network::Mainnet, node, &collector);
+        let mut pc = Buffer::new(
+            PeerConnector::new(config, Network::Mainnet, node, &collector),
+            1,
+        );
 
         let mut listener = TcpListener::bind(self.addr).await?;
 
         loop {
             let (tcp_stream, addr) = listener.accept().await?;
 
-            pc.ready().await?;
+            pc.ready()
-            let mut client = pc.call((tcp_stream, addr)).await?;
+                .await
+                .map_err(failure::Error::from_boxed_compat)?;
+            let mut client = pc
+                .call((tcp_stream, addr))
+                .await
+                .map_err(failure::Error::from_boxed_compat)?;
 
             let addrs = match client.call(Request::GetPeers).await? {
                 Response::Peers(addrs) => addrs,