Ported code to use futures_codec and tokio_util's Compat

Cargo.toml

@@ -14,7 +14,7 @@ async-trait = "^0.1"
 rand = "^0.7"
 bytes = "^0.5"
 tokio = { version = "^0.2", features = ["full", "tcp"] }
-tokio-util = { version = "^0.2", features = ["codec"] }
+tokio-util = { version = "0.3", features = ["compat"] }
 num-traits = "^0.2"
 enum-primitive-derive = "^0.1"
 dashmap = "^3.11"
@@ -23,14 +23,19 @@ uuid = { version = "^0.8", features = ["v4"] }
 regex = "1"
 lazy_static = "1"
 log = "0.4"
+futures_codec = "0.4"
 
 [dev-dependencies]
 chrono = "^0.4"
 criterion = "0.3"
 
-# [[bench]]
-# name = "pub_sub"
-# harness = false
+[lib]
+bench = false
+
+[[bench]]
+name = "pub_sub" 
+harness = false 
+bench = false # Don't actually benchmark this, until we fix it
 
 [[bench]]
 name = "req_rep"

benches/req_rep.rs

@@ -79,8 +79,8 @@ criterion_group! {
     name = benches;
     config = Criterion::default()
         .sample_size(128)
-        .measurement_time(Duration::from_secs(32))
-        .warm_up_time(Duration::from_secs(5));
+        .measurement_time(Duration::from_secs(30))
+        .warm_up_time(Duration::from_secs(10));
     targets = criterion_benchmark
 }
 criterion_main!(benches);

src/codec.rs

@@ -1,9 +1,9 @@
 use crate::error::ZmqError;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
+use futures_codec::{Decoder, Encoder};
 use std::collections::HashMap;
 use std::convert::TryFrom;
 use std::fmt::Display;
-use tokio_util::codec::{Decoder, Encoder};
 
 use crate::message::*;
 use crate::SocketType;

src/dealer_router.rs

@@ -3,10 +3,10 @@ use dashmap::DashMap;
 use futures::channel::{mpsc, oneshot};
 use futures::lock::Mutex;
 use futures::SinkExt;
+use futures_codec::Framed;
 use std::convert::TryInto;
 use std::sync::Arc;
 use tokio::net::TcpStream;
-use tokio_util::codec::Framed;
 
 use crate::codec::*;
 use crate::endpoint::{Endpoint, TryIntoEndpoint};

src/lib.rs

@@ -8,7 +8,7 @@ use futures::channel::{mpsc, oneshot};
 use std::convert::TryFrom;
 use std::fmt::{Debug, Display};
 
-use tokio_util::codec::Framed;
+use futures_codec::Framed;
 
 mod codec;
 mod dealer_router;

src/util.rs

@@ -9,6 +9,8 @@ use futures_util::future::FutureExt;
 use std::convert::{TryFrom, TryInto};
 use std::sync::Arc;
 use tokio::net::TcpStream;
+use tokio_util::compat::Compat;
+use tokio_util::compat::Tokio02AsyncReadCompatExt;
 use uuid::Uuid;
 
 #[derive(PartialEq, Eq, PartialOrd, Ord, Debug, Hash, Clone)]
@@ -87,7 +89,9 @@ pub fn sockets_compatible(one: SocketType, another: SocketType) -> bool {
     COMPATIBILITY_MATRIX[row_index * 11 + col_index] != 0
 }
 
-pub(crate) async fn greet_exchange(socket: &mut Framed<TcpStream, ZmqCodec>) -> ZmqResult<()> {
+pub(crate) async fn greet_exchange(
+    socket: &mut Framed<Compat<TcpStream>, ZmqCodec>,
+) -> ZmqResult<()> {
     socket
         .send(Message::Greeting(ZmqGreeting::default()))
         .await?;
@@ -104,7 +108,7 @@ pub(crate) async fn greet_exchange(socket: &mut Framed<TcpStream, ZmqCodec>) ->
 }
 
 pub(crate) async fn ready_exchange(
-    socket: &mut Framed<TcpStream, ZmqCodec>,
+    socket: &mut Framed<Compat<TcpStream>, ZmqCodec>,
     socket_type: SocketType,
 ) -> ZmqResult<PeerIdentity> {
     let ready = ZmqCommand::ready(socket_type);
@@ -143,7 +147,7 @@ pub(crate) async fn ready_exchange(
 }
 
 pub(crate) async fn peer_connected(socket: tokio::net::TcpStream, backend: Arc<dyn MultiPeer>) {
-    let mut raw_socket = Framed::new(socket, ZmqCodec::new());
+    let mut raw_socket = Framed::new(socket.compat(), ZmqCodec::new());
 
     greet_exchange(&mut raw_socket)
         .await