chore(deps): replace futures with futures-util (#281)

Author: paolobarbolini

Cargo.toml

@@ -137,8 +137,8 @@ hyper-rustls = { version = "0.27.3", default-features = false, features = [
     "tls12",
 ], optional = true }
 url = "2.1.1"
-futures = "0.3.5"
-futures-channel = "0.3.30"
+futures-util = { version = "0.3.5", default-features = false, features = ["sink", "io"] }
+futures-channel = { version = "0.3.30", features = ["sink"] }
 static_assertions = "1.1"
 sealed = "0.6"
 lz4_flex = { version = "0.11.3", default-features = false, features = [

benches/common.rs

@@ -12,7 +12,7 @@ use std::{
 
 use bytes::Bytes;
 use clickhouse::error::Result;
-use futures::stream::StreamExt;
+use futures_util::stream::StreamExt;
 use http_body_util::BodyExt;
 use hyper::{
     body::{Body, Incoming},

benches/mocked_select.rs

@@ -5,7 +5,7 @@ use clickhouse::{
 };
 use clickhouse_types::{Column, DataTypeNode};
 use criterion::{criterion_group, criterion_main, Criterion, Throughput};
-use futures::stream::{self, StreamExt as _};
+use futures_util::stream::{self, StreamExt as _};
 use http_body_util::StreamBody;
 use hyper::{
     body::{Body, Frame, Incoming},

examples/stream_into_file.rs

@@ -45,10 +45,10 @@ async fn cursor_next(filename: &str) {
     }
 }
 
-// Pattern 3: use the `futures::(Try)StreamExt` traits.
+// Pattern 3: use the `futures_util::(Try)StreamExt` traits.
 #[cfg(feature = "futures03")]
 async fn futures03_stream(filename: &str) {
-    use futures::TryStreamExt;
+    use futures_util::TryStreamExt;
 
     let mut cursor = query(NUMBERS);
     let mut file = File::create(filename).await.unwrap();

src/compression/lz4.rs

@@ -1,11 +1,11 @@
 use std::{
     pin::Pin,
-    task::{Context, Poll},
+    task::{ready, Context, Poll},
 };
 
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 use cityhash_rs::cityhash_102_128;
-use futures::{ready, stream::Stream};
+use futures_util::stream::Stream;
 use lz4_flex::block;
 
 use crate::{
@@ -182,7 +182,7 @@ pub(crate) fn compress(uncompressed: &[u8]) -> Result<Bytes> {
 
 #[tokio::test]
 async fn it_decompresses() {
-    use futures::stream::{self, TryStreamExt};
+    use futures_util::stream::{self, TryStreamExt};
 
     let expected = vec![
         1u8, 0, 2, 255, 255, 255, 255, 0, 1, 1, 1, 115, 6, 83, 116, 114, 105, 110, 103, 3, 97, 98,

src/cursors/bytes.rs

@@ -17,8 +17,9 @@ use tokio::io::{AsyncBufRead, AsyncRead, ReadBuf};
 /// Additionally to [`BytesCursor::next`] and [`BytesCursor::collect`],
 /// this cursor implements:
 /// * [`AsyncRead`] and [`AsyncBufRead`] for `tokio`-based ecosystem.
-/// * [`futures::Stream`], [`futures::AsyncRead`] and [`futures::AsyncBufRead`]
-///   for `futures`-based ecosystem. (requires the `futures03` feature)
+/// * [`futures_util::Stream`], [`futures_util::AsyncRead`] and
+///   [`futures_util::AsyncBufRead`] for `futures`-based ecosystem.
+///   (requires the `futures03` feature)
 ///
 /// For instance, if the requested format emits each row on a newline
 /// (e.g. `JSONEachRow`, `CSV`, `TSV`, etc.), the cursor can be read line by
@@ -167,7 +168,7 @@ impl AsyncBufRead for BytesCursor {
 }
 
 #[cfg(feature = "futures03")]
-impl futures::AsyncRead for BytesCursor {
+impl futures_util::AsyncRead for BytesCursor {
     #[inline]
     fn poll_read(
         self: Pin<&mut Self>,
@@ -181,7 +182,7 @@ impl futures::AsyncRead for BytesCursor {
 }
 
 #[cfg(feature = "futures03")]
-impl futures::AsyncBufRead for BytesCursor {
+impl futures_util::AsyncBufRead for BytesCursor {
     #[inline]
     fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<IoResult<&[u8]>> {
         AsyncBufRead::poll_fill_buf(self, cx)
@@ -194,7 +195,7 @@ impl futures::AsyncBufRead for BytesCursor {
 }
 
 #[cfg(feature = "futures03")]
-impl futures::stream::Stream for BytesCursor {
+impl futures_util::stream::Stream for BytesCursor {
     type Item = crate::error::Result<bytes::Bytes>;
 
     #[inline]
@@ -209,7 +210,7 @@ impl futures::stream::Stream for BytesCursor {
 }
 
 #[cfg(feature = "futures03")]
-impl futures::stream::FusedStream for BytesCursor {
+impl futures_util::stream::FusedStream for BytesCursor {
     #[inline]
     fn is_terminated(&self) -> bool {
         self.bytes.is_empty() && self.raw.is_terminated()

src/cursors/raw.rs

@@ -3,7 +3,7 @@ use crate::{
     response::{Chunks, Response, ResponseFuture},
 };
 use bytes::Bytes;
-use futures::Stream;
+use futures_util::Stream;
 use std::{
     pin::pin,
     task::{ready, Context, Poll},

src/request_body.rs

@@ -6,8 +6,8 @@ use std::{
 };
 
 use bytes::Bytes;
-use futures::{SinkExt, Stream};
 use futures_channel::mpsc;
+use futures_util::{SinkExt, Stream};
 use hyper::body::{Body, Frame, SizeHint};
 
 // === RequestBody ===

src/response.rs

@@ -1,15 +1,12 @@
 use std::{
-    future::Future,
-    pin::Pin,
+    future::{self, Future},
+    pin::{pin, Pin},
     task::{Context, Poll},
 };
 
 use bstr::ByteSlice;
 use bytes::{BufMut, Bytes};
-use futures::{
-    future,
-    stream::{self, Stream, TryStreamExt},
-};
+use futures_util::stream::{self, Stream, TryStreamExt};
 use http_body_util::BodyExt as _;
 use hyper::{
     body::{Body as _, Incoming},
@@ -125,7 +122,7 @@ async fn collect_bad_response(
 }
 
 async fn collect_bytes(stream: impl Stream<Item = Result<Bytes>>) -> Result<Bytes> {
-    futures::pin_mut!(stream);
+    let mut stream = pin!(stream);
 
     let mut bytes = Vec::new();
 

src/test/handlers.rs

@@ -1,7 +1,7 @@
 use std::marker::PhantomData;
 
 use bytes::Bytes;
-use futures::channel::oneshot;
+use futures_channel::oneshot;
 use hyper::{Request, Response, StatusCode};
 use sealed::sealed;
 use serde::Serialize;