replace futures crate api with standard library (#1461)

* replace futures crate api with standard library

Signed-off-by: tottoto <[email protected]>

* Move pin value in examples to independent sentence

Signed-off-by: tottoto <[email protected]>

---------

Signed-off-by: tottoto <[email protected]>

Author: tottoto

clippy.toml

@@ -1 +1,8 @@
 disallowed-names = []
+disallowed-macros = [
+  "futures::ready",             # use instead `std::task::ready`
+  "futures::pin_mut",           # use instead `std::pin::pin`
+]
+disallowed-methods = [
+  "futures::future::ready",     # use instead `std::future::ready`
+]

examples/event_watcher.rs

@@ -1,4 +1,6 @@
-use futures::{pin_mut, TryStreamExt};
+use std::pin::pin;
+
+use futures::TryStreamExt;
 use k8s_openapi::{
     api::{core::v1::ObjectReference, events::v1::Event},
     apimachinery::pkg::apis::meta::v1::Time,
@@ -37,7 +39,7 @@ async fn main() -> anyhow::Result<()> {
         }
     }
     let event_stream = watcher(events, conf).default_backoff().applied_objects();
-    pin_mut!(event_stream);
+    let mut event_stream = pin!(event_stream);
 
     println!("{0:<6} {1:<15} {2:<55} {3}", "AGE", "REASON", "OBJECT", "MESSAGE");
     while let Some(ev) = event_stream.try_next().await? {

examples/node_reflector.rs

@@ -1,4 +1,6 @@
-use futures::{pin_mut, TryStreamExt};
+use std::pin::pin;
+
+use futures::TryStreamExt;
 use k8s_openapi::api::core::v1::Node;
 use kube::{
     api::{Api, ResourceExt},
@@ -23,6 +25,7 @@ async fn main() -> anyhow::Result<()> {
         .reflect(writer)
         .applied_objects()
         .predicate_filter(predicates::labels.combine(predicates::annotations)); // NB: requires an unstable feature
+    let mut stream = pin!(stream);
 
     // Periodically read our state in the background
     tokio::spawn(async move {
@@ -35,7 +38,6 @@ async fn main() -> anyhow::Result<()> {
     });
 
     // Log applied events with changes from the reflector
-    pin_mut!(stream);
     while let Some(node) = stream.try_next().await? {
         info!("saw node {} with new labels/annots", node.name_any());
     }

examples/node_watcher.rs

@@ -1,4 +1,6 @@
-use futures::{pin_mut, TryStreamExt};
+use std::pin::pin;
+
+use futures::TryStreamExt;
 use k8s_openapi::api::core::v1::{Event, Node};
 use kube::{
     api::{Api, ListParams, ResourceExt},
@@ -21,8 +23,8 @@ async fn main() -> anyhow::Result<()> {
         watcher::Config::default()
     };
     let obs = watcher(nodes, wc).default_backoff().applied_objects();
+    let mut obs = pin!(obs);
 
-    pin_mut!(obs);
     while let Some(n) = obs.try_next().await? {
         check_for_node_failures(&client, n).await?;
     }

examples/pod_reflector.rs

@@ -1,3 +1,5 @@
+use std::pin::pin;
+
 use futures::TryStreamExt;
 use k8s_openapi::api::core::v1::Pod;
 use kube::{
@@ -40,7 +42,7 @@ async fn main() -> anyhow::Result<()> {
         .reflect(writer)
         .applied_objects()
         .predicate_filter(predicates::resource_version); // NB: requires an unstable feature
-    futures::pin_mut!(stream);
+    let mut stream = pin!(stream);
 
     while let Some(pod) = stream.try_next().await? {
         info!("saw {}", pod.name_any());

kube-client/src/client/middleware/mod.rs

@@ -27,10 +27,9 @@ impl<S> Layer<S> for AuthLayer {
 mod tests {
     use super::*;
 
-    use std::{matches, sync::Arc};
+    use std::{matches, pin::pin, sync::Arc};
 
     use chrono::{Duration, Utc};
-    use futures::pin_mut;
     use http::{header::AUTHORIZATION, HeaderValue, Request, Response};
     use secrecy::SecretString;
     use tokio::sync::Mutex;
@@ -52,7 +51,7 @@ mod tests {
 
         let spawned = tokio::spawn(async move {
             // Receive the requests and respond
-            pin_mut!(handle);
+            let mut handle = pin!(handle);
             let (request, send) = handle.next_request().await.expect("service not called");
             assert_eq!(
                 request.headers().get(AUTHORIZATION).unwrap(),

kube-client/src/client/mod.rs

@@ -8,7 +8,7 @@
 //! The [`Client`] can also be used with [`Discovery`](crate::Discovery) to dynamically
 //! retrieve the resources served by the kubernetes API.
 use either::{Either, Left, Right};
-use futures::{self, AsyncBufRead, StreamExt, TryStream, TryStreamExt};
+use futures::{AsyncBufRead, StreamExt, TryStream, TryStreamExt};
 use http::{self, Request, Response};
 use http_body_util::BodyExt;
 #[cfg(feature = "ws")] use hyper_util::rt::TokioIo;
@@ -492,9 +492,10 @@ impl TryFrom<Config> for Client {
 
 #[cfg(test)]
 mod tests {
+    use std::pin::pin;
+
     use crate::{client::Body, Api, Client};
 
-    use futures::pin_mut;
     use http::{Request, Response};
     use k8s_openapi::api::core::v1::Pod;
     use tower_test::mock;
@@ -511,7 +512,7 @@ mod tests {
         let (mock_service, handle) = mock::pair::<Request<Body>, Response<Body>>();
         let spawned = tokio::spawn(async move {
             // Receive a request for pod and respond with some data
-            pin_mut!(handle);
+            let mut handle = pin!(handle);
             let (request, send) = handle.next_request().await.expect("service not called");
             assert_eq!(request.method(), http::Method::GET);
             assert_eq!(request.uri().to_string(), "/api/v1/namespaces/default/pods/test");

kube-runtime/src/controller/future_hash_map.rs

@@ -77,10 +77,10 @@ where
 
 #[cfg(test)]
 mod tests {
-    use std::task::Poll;
+    use std::{future, task::Poll};
 
     use super::FutureHashMap;
-    use futures::{channel::mpsc, future, poll, StreamExt};
+    use futures::{channel::mpsc, poll, StreamExt};
 
     #[tokio::test]
     async fn fhm_should_forward_all_values_and_shut_down() {

kube-runtime/src/controller/mod.rs

@@ -16,16 +16,17 @@ use derivative::Derivative;
 use futures::{
     channel,
     future::{self, BoxFuture},
-    ready, stream, Future, FutureExt, Stream, StreamExt, TryFuture, TryFutureExt, TryStream, TryStreamExt,
+    stream, FutureExt, Stream, StreamExt, TryFuture, TryFutureExt, TryStream, TryStreamExt,
 };
 use kube_client::api::{Api, DynamicObject, Resource};
 use pin_project::pin_project;
 use serde::de::DeserializeOwned;
 use std::{
     fmt::{Debug, Display},
+    future::Future,
     hash::Hash,
     sync::Arc,
-    task::Poll,
+    task::{ready, Poll},
     time::Duration,
 };
 use stream::BoxStream;
@@ -326,7 +327,8 @@ where
                                 .instrument(reconciler_span)
                                 .left_future()
                         }
-                        None => future::err(Error::ObjectNotFound(request.obj_ref.erase())).right_future(),
+                        None => std::future::ready(Err(Error::ObjectNotFound(request.obj_ref.erase())))
+                            .right_future(),
                     }
                 },
             )
@@ -1155,7 +1157,7 @@ where
     /// use kube::{Api, Client, ResourceExt};
     /// use kube_runtime::{
     ///     controller::{Controller, Action},
-    ///     watcher,  
+    ///     watcher,
     /// };
     /// use std::{convert::Infallible, sync::Arc};
     /// Controller::new(
@@ -1274,7 +1276,7 @@ where
 
 #[cfg(test)]
 mod tests {
-    use std::{convert::Infallible, sync::Arc, time::Duration};
+    use std::{convert::Infallible, pin::pin, sync::Arc, time::Duration};
 
     use super::{Action, APPLIER_REQUEUE_BUF_SIZE};
     use crate::{
@@ -1283,7 +1285,7 @@ mod tests {
         watcher::{self, metadata_watcher, watcher, Event},
         Config, Controller,
     };
-    use futures::{pin_mut, Stream, StreamExt, TryStreamExt};
+    use futures::{Stream, StreamExt, TryStreamExt};
     use k8s_openapi::api::core::v1::ConfigMap;
     use kube_client::{core::ObjectMeta, Api, Resource};
     use serde::de::DeserializeOwned;
@@ -1348,7 +1350,7 @@ mod tests {
 
         let (queue_tx, queue_rx) = futures::channel::mpsc::unbounded::<ObjectRef<ConfigMap>>();
         let (store_rx, mut store_tx) = reflector::store();
-        let applier = applier(
+        let mut applier = pin!(applier(
             |obj, _| {
                 Box::pin(async move {
                     // Try to flood the rescheduling buffer buffer by just putting it back in the queue immediately
@@ -1361,8 +1363,7 @@ mod tests {
             store_rx,
             queue_rx.map(Result::<_, Infallible>::Ok),
             Config::default(),
-        );
-        pin_mut!(applier);
+        ));
         for i in 0..items {
             let obj = ConfigMap {
                 metadata: ObjectMeta {

kube-runtime/src/controller/runner.rs

@@ -1,9 +1,10 @@
 use super::future_hash_map::FutureHashMap;
 use crate::scheduler::{ScheduleRequest, Scheduler};
-use futures::{future, Future, FutureExt, Stream, StreamExt};
+use futures::{FutureExt, Stream, StreamExt};
 use pin_project::pin_project;
 use std::{
     convert::Infallible,
+    future::{self, Future},
     hash::Hash,
     pin::Pin,
     task::{Context, Poll},
@@ -29,7 +30,7 @@ pub struct Runner<T, R, F, MkF, Ready = future::Ready<Result<(), Infallible>>> {
     run_msg: MkF,
     slots: FutureHashMap<T, F>,
     #[pin]
-    ready_to_execute_after: future::Fuse<Ready>,
+    ready_to_execute_after: futures::future::Fuse<Ready>,
     is_ready_to_execute: bool,
     stopped: bool,
     max_concurrent_executions: u16,
@@ -163,8 +164,7 @@ mod tests {
     };
     use futures::{
         channel::{mpsc, oneshot},
-        future::{self},
-        poll, stream, Future, SinkExt, StreamExt, TryStreamExt,
+        future, poll, stream, Future, SinkExt, StreamExt, TryStreamExt,
     };
     use std::{
         cell::RefCell,