migrate to using plain futures crate

Cargo.toml

@@ -21,10 +21,7 @@ tcp-transport = []
 [dependencies]
 async-dispatcher = { version = "0.1", optional = true }
 thiserror = "1"
-futures-channel = { version = "0.3", features = ["sink"] }
-futures-io = "0.3"
-futures-task = "0.3"
-futures-util = { version = "0.3", features = ["sink"] }
+futures = "0.3"
 async-trait = "0.1"
 parking_lot = "0.12"
 rand = "0.8"

examples/task_worker.rs

@@ -1,6 +1,6 @@
 mod async_helpers;
 
-use futures_util::{select, FutureExt};
+use futures::{select, FutureExt};
 use std::io::Write;
 use std::{error::Error, time::Duration};
 use zeromq::{Socket, SocketRecv, SocketSend};

src/backend.rs

@@ -8,8 +8,8 @@ use crate::{
 use async_trait::async_trait;
 use crossbeam_queue::SegQueue;
 use dashmap::DashMap;
-use futures_channel::mpsc;
-use futures_util::SinkExt;
+use futures::channel::mpsc;
+use futures::SinkExt;
 use parking_lot::Mutex;
 
 use std::sync::Arc;

src/codec/framed.rs

@@ -1,7 +1,7 @@
 use crate::codec::ZmqCodec;
 
 use asynchronous_codec::{FramedRead, FramedWrite};
-use futures_io::{AsyncRead, AsyncWrite};
+use futures::{AsyncRead, AsyncWrite};
 
 // Enables us to have multiple bounds on the dyn trait in `InnerFramed`
 pub trait FrameableRead: AsyncRead + Unpin + Send + Sync {}

src/codec/mod.rs

@@ -16,9 +16,8 @@ pub(crate) use zmq_codec::ZmqCodec;
 
 use crate::message::ZmqMessage;
 use crate::{ZmqError, ZmqResult};
-
-use futures_task::noop_waker;
-use futures_util::Sink;
+use futures::task::noop_waker;
+use futures::Sink;
 
 use std::pin::Pin;
 use std::task::{Context, Poll};

src/dealer.rs

@@ -9,8 +9,8 @@ use crate::{
 };
 
 use async_trait::async_trait;
-use futures_channel::mpsc;
-use futures_util::StreamExt;
+use futures::channel::mpsc;
+use futures::StreamExt;
 
 use std::collections::hash_map::RandomState;
 use std::collections::HashMap;

src/error.rs

@@ -4,7 +4,7 @@ use crate::endpoint::EndpointError;
 use crate::task_handle::TaskError;
 use crate::ZmqMessage;
 
-use futures_channel::mpsc;
+use futures::channel::mpsc;
 use thiserror::Error;
 
 pub type ZmqResult<T> = Result<T, ZmqError>;

src/fair_queue.rs

@@ -1,5 +1,5 @@
-use futures_task::{waker_ref, ArcWake};
-use futures_util::Stream;
+use futures::task::{waker_ref, ArcWake};
+use futures::Stream;
 use parking_lot::Mutex;
 
 use std::cmp::Ordering;
@@ -24,10 +24,9 @@ impl<S, K: Clone + Eq + Hash> QueueInner<S, K> {
             priority: self.counter.fetch_add(1, atomic::Ordering::Relaxed),
             key: k,
         });
-        match &self.waker {
-            Some(w) => w.wake_by_ref(),
-            None => (),
-        };
+        if let Some(w) = &self.waker {
+            w.wake_by_ref();
+        }
     }
 
     pub fn remove(&mut self, k: &K) {
@@ -164,7 +163,7 @@ impl<S, K: Clone> FairQueue<S, K> {
 mod test {
     use crate::async_rt;
     use crate::fair_queue::FairQueue;
-    use futures_util::{stream, StreamExt};
+    use futures::{stream, StreamExt};
 
     #[async_rt::test]
     async fn test_fair_queue_ready() {

src/lib.rs

@@ -43,8 +43,8 @@ use util::PeerIdentity;
 
 use async_trait::async_trait;
 use asynchronous_codec::FramedWrite;
-use futures_channel::mpsc;
-use futures_util::{select, FutureExt};
+use futures::channel::mpsc;
+use futures::{select, FutureExt};
 use parking_lot::Mutex;
 
 use std::collections::HashMap;
@@ -185,8 +185,8 @@ impl SocketOptions {
 pub trait MultiPeerBackend: SocketBackend {
     /// This should not be public..
     /// Find a better way of doing this
-
     async fn peer_connected(self: Arc<Self>, peer_id: &PeerIdentity, io: FramedIo);
+
     fn peer_disconnected(&self, peer_id: &PeerIdentity);
 }
 
@@ -211,6 +211,7 @@ pub trait SocketSend {
 /// in [proxy] function as a capture parameter
 pub trait CaptureSocket: SocketSend {}
 
+#[allow(clippy::empty_line_after_outer_attr)]
 #[async_trait]
 pub trait Socket: Sized + Send {
     fn new() -> Self {
@@ -333,11 +334,11 @@ pub trait Socket: Sized + Send {
     /// # Errors
     /// May give a `ZmqError::NoSuchConnection` if `endpoint` isn't connected.
     /// May also give any other zmq errors encountered when attempting to
-    /// disconnect.
+    /// disconnect
     // TODO: async fn disconnect(&mut self, endpoint: impl TryIntoEndpoint + 'async_trait) ->
     // ZmqResult<()>;
 
-    /// Disconnects all connecttions, blocking until finished.
+    /// Disconnects all connections, blocking until finished.
     // TODO: async fn disconnect_all(&mut self) -> ZmqResult<()>;
 
     /// Closes the socket, blocking until all associated binds are closed.

src/pub.rs

@@ -11,8 +11,8 @@ use crate::{
 
 use async_trait::async_trait;
 use dashmap::DashMap;
-use futures_channel::{mpsc, oneshot};
-use futures_util::{select, FutureExt, StreamExt};
+use futures::channel::{mpsc, oneshot};
+use futures::{select, FutureExt, StreamExt};
 use parking_lot::Mutex;
 
 use std::collections::HashMap;

src/pull.rs

@@ -9,8 +9,8 @@ use crate::{
 };
 
 use async_trait::async_trait;
-use futures_channel::mpsc;
-use futures_util::StreamExt;
+use futures::channel::mpsc;
+use futures::StreamExt;
 
 use std::collections::hash_map::RandomState;
 use std::collections::HashMap;

src/push.rs

@@ -7,7 +7,7 @@ use crate::{
 };
 
 use async_trait::async_trait;
-use futures_channel::mpsc;
+use futures::channel::mpsc;
 
 use std::collections::hash_map::RandomState;
 use std::collections::HashMap;

src/rep.rs

@@ -8,7 +8,7 @@ use crate::{SocketType, ZmqResult};
 
 use async_trait::async_trait;
 use dashmap::DashMap;
-use futures_util::{SinkExt, StreamExt};
+use futures::{SinkExt, StreamExt};
 use parking_lot::Mutex;
 
 use std::collections::HashMap;

src/req.rs

@@ -10,7 +10,7 @@ use async_trait::async_trait;
 use bytes::Bytes;
 use crossbeam_queue::SegQueue;
 use dashmap::DashMap;
-use futures_util::{SinkExt, StreamExt};
+use futures::{SinkExt, StreamExt};
 
 use std::collections::HashMap;
 use std::sync::Arc;

src/router.rs

@@ -10,8 +10,8 @@ use crate::{MultiPeerBackend, SocketEvent, SocketOptions, SocketRecv, SocketSend
 use crate::{Socket, SocketBackend};
 
 use async_trait::async_trait;
-use futures_channel::mpsc;
-use futures_util::{SinkExt, StreamExt};
+use futures::channel::mpsc;
+use futures::{SinkExt, StreamExt};
 
 use std::collections::HashMap;
 use std::convert::TryInto;

src/sub.rs

@@ -15,8 +15,8 @@ use async_trait::async_trait;
 use bytes::{BufMut, BytesMut};
 use crossbeam_queue::SegQueue;
 use dashmap::DashMap;
-use futures_channel::mpsc;
-use futures_util::{SinkExt, StreamExt};
+use futures::channel::mpsc;
+use futures::{SinkExt, StreamExt};
 use parking_lot::Mutex;
 
 use std::collections::{HashMap, HashSet};

src/task_handle.rs

@@ -1,7 +1,7 @@
 use crate::async_rt;
 use crate::error::{ZmqError, ZmqResult};
 
-use futures_channel::oneshot;
+use futures::channel::oneshot;
 use thiserror::Error;
 
 #[derive(Error, Debug)]

src/transport/ipc.rs

@@ -12,8 +12,8 @@ use crate::endpoint::Endpoint;
 use crate::task_handle::TaskHandle;
 use crate::ZmqResult;
 
-use futures_channel::oneshot;
-use futures_util::{select, FutureExt};
+use futures::channel::oneshot;
+use futures::{select, FutureExt};
 
 use std::path::Path;
 

src/transport/mod.rs

@@ -104,9 +104,9 @@ where
 #[cfg(any(feature = "async-std-runtime", feature = "async-dispatcher-runtime"))]
 fn make_framed<T>(stream: T) -> FramedIo
 where
-    T: futures_io::AsyncRead + futures_io::AsyncWrite + Send + Sync + 'static,
+    T: futures::AsyncRead + futures::AsyncWrite + Send + Sync + 'static,
 {
-    use futures_util::AsyncReadExt;
+    use futures::AsyncReadExt;
     let (read, write) = stream.split();
     FramedIo::new(Box::new(read), Box::new(write))
 }

src/transport/tcp.rs

@@ -12,7 +12,7 @@ use crate::endpoint::{Endpoint, Host, Port};
 use crate::task_handle::TaskHandle;
 use crate::ZmqResult;
 
-use futures_util::{select, FutureExt};
+use futures::{select, FutureExt};
 
 pub(crate) async fn connect(host: &Host, port: Port) -> ZmqResult<(FramedIo, Endpoint)> {
     let raw_socket = TcpStream::connect((host.to_string().as_str(), port)).await?;
@@ -35,7 +35,7 @@ where
 {
     let listener = TcpListener::bind((host.to_string().as_str(), port)).await?;
     let resolved_addr = listener.local_addr()?;
-    let (stop_channel, stop_callback) = futures_channel::oneshot::channel::<()>();
+    let (stop_channel, stop_callback) = futures::channel::oneshot::channel::<()>();
     let task_handle = async_rt::task::spawn(async move {
         let mut stop_callback = stop_callback.fuse();
         loop {

src/util.rs

@@ -3,7 +3,7 @@ use crate::*;
 
 use asynchronous_codec::FramedRead;
 use bytes::Bytes;
-use futures_util::{SinkExt, StreamExt};
+use futures::{SinkExt, StreamExt};
 use rand::Rng;
 
 use std::convert::{TryFrom, TryInto};

tests/pub_sub.rs

@@ -5,8 +5,8 @@ mod test {
     use zeromq::ZmqMessage;
     use zeromq::__async_rt as async_rt;
 
-    use futures_channel::{mpsc, oneshot};
-    use futures_util::{SinkExt, StreamExt};
+    use futures::channel::{mpsc, oneshot};
+    use futures::{SinkExt, StreamExt};
     use std::time::Duration;
 
     #[async_rt::test]
@@ -102,6 +102,6 @@ mod test {
             "ipc://asdf.sock",
             "ipc://anothersocket-asdf",
         ];
-        futures_util::future::join_all(addrs.into_iter().map(helper)).await;
+        futures::future::join_all(addrs.into_iter().map(helper)).await;
     }
 }

tests/req_rep.rs

@@ -3,7 +3,7 @@ mod helpers;
 use zeromq::__async_rt as async_rt;
 use zeromq::prelude::*;
 
-use futures_util::StreamExt;
+use futures::StreamExt;
 use std::error::Error;
 use std::time::Duration;
 

tests/req_router_dealer_rep.rs

@@ -7,7 +7,7 @@ mod test {
     use zeromq::__async_rt as async_rt;
     use zeromq::prelude::*;
 
-    use futures_util::StreamExt;
+    use futures::StreamExt;
     use std::error::Error;
     use std::time::Duration;