Return a Stream<Item=Result<T>> from request_events

Cargo.toml

@@ -29,6 +29,7 @@ time = "0.1.42"
 either = "1.5.3"
 thiserror = "1.0.5"
 futures-timer = "2.0.0"
+futures = "0.3.1"
 
 [features]
 default = []

examples/event_informer.rs

@@ -1,11 +1,14 @@
-#[macro_use] extern crate log;
+#[macro_use]
+extern crate log;
 use kube::{
-    api::{Api, Informer, WatchEvent},
     api::v1Event,
+    api::{Api, Informer, WatchEvent},
     client::APIClient,
     config,
 };
 
+use futures::StreamExt;
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     std::env::set_var("RUST_LOG", "info,kube=trace");
@@ -14,13 +17,13 @@ async fn main() -> anyhow::Result<()> {
     let client = APIClient::new(config);
 
     let events = Api::v1Event(client);
-    let ei = Informer::new(events)
-        .init().await?;
+    let ei = Informer::new(events).init().await?;
 
     loop {
-        ei.poll().await?;
+        let mut events = ei.poll().await?.boxed();
 
-        while let Some(event) = ei.pop() {
+        while let Some(event) = events.next().await {
+            let event = event?;
             handle_events(event)?;
         }
     }
@@ -31,13 +34,13 @@ fn handle_events(ev: WatchEvent<v1Event>) -> anyhow::Result<()> {
     match ev {
         WatchEvent::Added(o) => {
             info!("New Event: {}, {}", o.type_, o.message);
-        },
+        }
         WatchEvent::Modified(o) => {
             info!("Modified Event: {}", o.reason);
-        },
+        }
         WatchEvent::Deleted(o) => {
             info!("Deleted Event: {}", o.message);
-        },
+        }
         WatchEvent::Error(e) => {
             warn!("Error event: {:?}", e);
         }

examples/job_openapi.rs

@@ -1,9 +1,11 @@
-#[macro_use] extern crate log;
+#[macro_use]
+extern crate log;
+use futures::StreamExt;
 use serde_json::json;
 
 use kube::{
-    api::{Api, PostParams, DeleteParams, ListParams, WatchEvent},
-    client::{APIClient},
+    api::{Api, DeleteParams, ListParams, PostParams, WatchEvent},
+    client::APIClient,
     config,
 };
 
@@ -45,36 +47,28 @@ async fn main() -> anyhow::Result<()> {
 
     // See if it ran to completion
     let lp = ListParams::default();
-    jobs.watch(&lp, "").await.and_then(|res| {
-        for status in res {
-            match status {
-                WatchEvent::Added(s) => {
-                    info!("Added {}", s.metadata.name);
-                },
-                WatchEvent::Modified(s) => {
-                    let current_status = s.status.clone().expect("Status is missing");
-                    current_status.completion_time.and_then(|_| {
-                        info!("Modified: {} is complete", s.metadata.name);
-                        Some(())
-                    }).or_else(|| {
-                        info!("Modified: {} is running", s.metadata.name);
-                        Some(())
-                    });
-                },
-                WatchEvent::Deleted(s) => {
-                    info!("Deleted {}", s.metadata.name);
-                }
-                WatchEvent::Error(s) => {
-                    error!("{}", s);
+    let mut stream = jobs.watch(&lp, "").await?.boxed();
+
+    while let Some(status) = stream.next().await {
+        match status {
+            WatchEvent::Added(s) => info!("Added {}", s.metadata.name),
+            WatchEvent::Modified(s) => {
+                let current_status = s.status.clone().expect("Status is missing");
+                match current_status.completion_time {
+                    Some(_) => info!("Modified: {} is complete", s.metadata.name),
+                    _ => info!("Modified: {} is running", s.metadata.name),
                 }
             }
+            WatchEvent::Deleted(s) => info!("Deleted {}", s.metadata.name),
+            WatchEvent::Error(s) => error!("{}", s),
         }
-        Ok(())
-    }).expect("Failed to watch");
+    }
 
     // Clean up the old job record..
     info!("Deleting the job record.");
     let dp = DeleteParams::default();
-    jobs.delete("empty-job", &dp).await.expect("failed to delete job");
+    jobs.delete("empty-job", &dp)
+        .await
+        .expect("failed to delete job");
     Ok(())
 }

examples/node_informer.rs

@@ -1,10 +1,12 @@
-#[macro_use] extern crate log;
+#[macro_use]
+extern crate log;
+use futures::StreamExt;
+use k8s_openapi::api::core::v1::{NodeSpec, NodeStatus};
 use kube::{
-    api::{RawApi, Api, v1Event, Informer, ListParams, WatchEvent, Object},
+    api::{v1Event, Api, Informer, ListParams, Object, RawApi, WatchEvent},
     client::APIClient,
     config,
 };
-use k8s_openapi::api::core::v1::{NodeSpec, NodeStatus};
 
 type Node = Object<NodeSpec, NodeStatus>;
 type Event = v1Event; // snowflake obj
@@ -20,12 +22,14 @@ async fn main() -> anyhow::Result<()> {
     let events = Api::v1Event(client.clone());
     let ni = Informer::raw(client.clone(), nodes)
         .labels("beta.kubernetes.io/os=linux")
-        .init().await?;
+        .init()
+        .await?;
 
     loop {
-        ni.poll().await?;
+        let mut nodes = ni.poll().await?.boxed();
 
-        while let Some(ne) = ni.pop() {
+        while let Some(ne) = nodes.next().await {
+            let ne = ne?;
             handle_nodes(&events, ne).await?;
         }
     }
@@ -36,19 +40,30 @@ async fn handle_nodes(events: &Api<Event>, ne: WatchEvent<Node>) -> anyhow::Resu
     match ne {
         WatchEvent::Added(o) => {
             info!("New Node: {}", o.spec.provider_id.unwrap());
-        },
+        }
         WatchEvent::Modified(o) => {
             // Nodes often modify a lot - only print broken nodes
             if let Some(true) = o.spec.unschedulable {
-                let failed = o.status.unwrap().conditions.unwrap().into_iter().filter(|c| {
-                    // In a failed state either some of the extra conditions are not False
-                    // Or the Ready state is False
-                    (c.status == "True" && c.type_ != "Ready") ||
-                    (c.status == "False" &&  c.type_ == "Ready")
-                }).map(|c| c.message).collect::<Vec<_>>(); // failed statuses
+                let failed = o
+                    .status
+                    .unwrap()
+                    .conditions
+                    .unwrap()
+                    .into_iter()
+                    .filter(|c| {
+                        // In a failed state either some of the extra conditions are not False
+                        // Or the Ready state is False
+                        (c.status == "True" && c.type_ != "Ready")
+                            || (c.status == "False" && c.type_ == "Ready")
+                    })
+                    .map(|c| c.message)
+                    .collect::<Vec<_>>(); // failed statuses
                 warn!("Unschedulable Node: {}, ({:?})", o.metadata.name, failed);
                 // Find events related to this node
-                let sel = format!("involvedObject.kind=Node,involvedObject.name={}", o.metadata.name);
+                let sel = format!(
+                    "involvedObject.kind=Node,involvedObject.name={}",
+                    o.metadata.name
+                );
                 let opts = ListParams {
                     field_selector: Some(sel),
                     ..Default::default()
@@ -61,14 +76,16 @@ async fn handle_nodes(events: &Api<Event>, ne: WatchEvent<Node>) -> anyhow::Resu
                 // Turn up logging above to see
                 debug!("Normal node: {}", o.metadata.name);
             }
-        },
+        }
         WatchEvent::Deleted(o) => {
-            warn!("Deleted node: {} ({:?}) running {:?} with labels: {:?}",
-                o.metadata.name, o.spec.provider_id.unwrap(),
+            warn!(
+                "Deleted node: {} ({:?}) running {:?} with labels: {:?}",
+                o.metadata.name,
+                o.spec.provider_id.unwrap(),
                 o.status.unwrap().conditions.unwrap(),
                 o.metadata.labels,
             );
-        },
+        }
         WatchEvent::Error(e) => {
             warn!("Error event: {:?}", e);
         }

examples/pod_informer.rs

@@ -1,11 +1,13 @@
-#[macro_use] extern crate log;
-use std::env;
+#[macro_use]
+extern crate log;
+use futures::StreamExt;
+use k8s_openapi::api::core::v1::{PodSpec, PodStatus};
 use kube::{
-    api::{Api, Informer, WatchEvent, Object},
+    api::{Api, Informer, Object, WatchEvent},
     client::APIClient,
     config,
 };
-use k8s_openapi::api::core::v1::{PodSpec, PodStatus};
+use std::env;
 type Pod = Object<PodSpec, PodStatus>;
 
 #[tokio::main]
@@ -22,10 +24,11 @@ async fn main() -> anyhow::Result<()> {
     // Here we both poll and reconcile based on events from the main thread
     // If you run this next to actix-web (say), spawn a thread and pass `inf` as app state
     loop {
-        inf.poll().await?;
+        let mut pods = inf.poll().await?.boxed();
 
         // Handle events one by one, draining the informer
-        while let Some(event) = inf.pop() {
+        while let Some(event) = pods.next().await {
+            let event = event?;
             handle_node(&resource, event)?;
         }
     }
@@ -35,17 +38,28 @@ async fn main() -> anyhow::Result<()> {
 fn handle_node(_pods: &Api<Pod>, ev: WatchEvent<Pod>) -> anyhow::Result<()> {
     match ev {
         WatchEvent::Added(o) => {
-            let containers = o.spec.containers.into_iter().map(|c| c.name).collect::<Vec<_>>();
-            info!("Added Pod: {} (containers={:?})", o.metadata.name, containers);
-        },
+            let containers = o
+                .spec
+                .containers
+                .into_iter()
+                .map(|c| c.name)
+                .collect::<Vec<_>>();
+            info!(
+                "Added Pod: {} (containers={:?})",
+                o.metadata.name, containers
+            );
+        }
         WatchEvent::Modified(o) => {
             let phase = o.status.unwrap().phase.unwrap();
             let owner = &o.metadata.ownerReferences[0];
-            info!("Modified Pod: {} (phase={}, owner={})", o.metadata.name, phase, owner.name);
-        },
+            info!(
+                "Modified Pod: {} (phase={}, owner={})",
+                o.metadata.name, phase, owner.name
+            );
+        }
         WatchEvent::Deleted(o) => {
             info!("Deleted Pod: {}", o.metadata.name);
-        },
+        }
         WatchEvent::Error(e) => {
             warn!("Error event: {:?}", e);
         }