test with watch

Cargo.toml

@@ -17,6 +17,7 @@ exclude = ["/target"]
 anyhow = "1.0.43"
 clap = "2.33.3"
 env_logger = "0.9.0"
+futures = "0.3.16"
 k8s-openapi = { version = "0.13.0", features = ["v1_20"], default-features = false }
 kube = "0.59.0"
 kube-runtime = "0.59.0"

src/main.rs

@@ -3,12 +3,13 @@ use std::time::Duration;
 use anyhow::{bail, Context, Result};
 use api::Pod;
 use clap::{App, Arg, ArgMatches};
+use futures::{StreamExt, TryStreamExt};
 use k8s_openapi::api::core::v1 as api;
 use k8s_openapi::apimachinery::pkg::api::resource::Quantity;
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
 use k8s_openapi::{Metadata, NamespaceResourceScope, Resource};
-use kube::api::ListParams;
-use kube::{Api, Client};
+use kube::api::{ListParams, WatchEvent};
+use kube::{Api, Client, ResourceExt};
 use log::{debug, error, info};
 use serde::{Deserialize, Serialize};
 
@@ -154,17 +155,21 @@ async fn main() -> Result<()> {
 
     // let producer_configs = matches.value_of("producer_configs");
 
-    let client = Client::try_default().await.context("building client");
-
-    if let Err(e) = client {
-        error!("{}", e);
-        bail!("Abort");
-    } else {
-        info!("Client ok")
+    let client = Client::try_default().await.context("building client")?;
+
+    let pods: Api<Pod> = Api::namespaced(client, namespace);
+    let mut stream = pods.watch(&ListParams::default(), "0").await?.boxed();
+    while let Some(event) = stream.try_next().await? {
+        match event {
+            WatchEvent::Added(pod) => info!("ADDED: {}", pod.name()),
+            WatchEvent::Modified(pod) => info!("UPDATED: {}", pod.name()),
+            WatchEvent::Deleted(pod) => info!("DELETED: {}", pod.name()),
+            WatchEvent::Error(e) => error!("ERROR: {} {} ({})", e.code, e.message, e.status),
+            _ => {}
+        };
     }
-
-    let pods: Api<Pod> = Api::namespaced(client.unwrap(), namespace);
-    debug!("entering loop");
+    Ok(())
+    /* debug!("entering loop");
     loop {
         let list_params = ListParams::default();
         let pods_list = pods.list(&list_params).await.context("listing pods")?;
@@ -176,5 +181,5 @@ async fn main() -> Result<()> {
         }
         //debug!("Sleeping for {} minute/s", update_interval);
         // std::thread::sleep(Duration::from_secs(update_interval * 60));
-    }
+    } */
 }