Trait WatchStreamExt 

pub trait WatchStreamExt: Stream {
    // Provided methods
    fn default_backoff(self) -> StreamBackoff<Self, DefaultBackoff>
       where Self: TryStream + Sized { ... }
    fn backoff<B>(self, b: B) -> StreamBackoff<Self, B>
       where B: Backoff,
             Self: TryStream + Sized { ... }
    fn applied_objects<K>(self) -> EventDecode<Self>
       where Self: Stream<Item = Result<Event<K>, Error>> + Sized { ... }
    fn touched_objects<K>(self) -> EventDecode<Self>
       where Self: Stream<Item = Result<Event<K>, Error>> + Sized { ... }
    fn modify<F, K>(self, f: F) -> EventModify<Self, F>
       where Self: Stream<Item = Result<Event<K>, Error>> + Sized,
             F: FnMut(&mut K) { ... }
    fn predicate_filter<K, P>(
        self,
        predicate: P,
        config: PredicateConfig,
    ) -> PredicateFilter<Self, K, P>
       where Self: Stream<Item = Result<K, Error>> + Sized,
             K: Resource + 'static,
             P: Predicate<K> + 'static { ... }
    fn reflect<K>(self, writer: Writer<K>) -> Reflect<Self, K>
       where Self: Stream<Item = Result<Event<K>>> + Sized,
             K: Resource + Clone + 'static,
             K::DynamicType: Eq + Hash + Clone { ... }
}

Extension trait for streams returned by watcher or reflector

Provided Methods

fn default_backoff(self) -> StreamBackoff<Self, DefaultBackoff>

where Self: TryStream + Sized,

Apply the DefaultBackoff watcher Backoff policy

This is recommended for controllers that want to play nicely with the apiserver.

fn backoff<B>(self, b: B) -> StreamBackoff<Self, B>

where B: Backoff, Self: TryStream + Sized,

Apply a specific Backoff policy to a Stream using StreamBackoff

fn applied_objects<K>(self) -> EventDecode<Self>

where Self: Stream<Item = Result<Event<K>, Error>> + Sized,

Decode a watcher() stream into a stream of applied objects

All Added/Modified events are passed through, and critical errors bubble up.

fn touched_objects<K>(self) -> EventDecode<Self>

where Self: Stream<Item = Result<Event<K>, Error>> + Sized,

Decode a watcher() stream into a stream of touched objects

All Added/Modified/Deleted events are passed through, and critical errors bubble up. Only the objects are retained; the original event type is discarded. An object from a delete event need not have a deletion timestamp or a changed generation. A subsequent predicate_filter can therefore suppress it if the selected properties are unchanged. See deletion handling.

fn modify<F, K>(self, f: F) -> EventModify<Self, F>

where Self: Stream<Item = Result<Event<K>, Error>> + Sized, F: FnMut(&mut K),

Modify elements of a watcher() stream.

Calls watcher::Event::modify() on every element. Stream shorthand for stream.map_ok(|event| { event.modify(f) }).

let deploys: Api<Deployment> = Api::all(client);
let mut truncated_deploy_stream = pin!(watcher(deploys, watcher::Config::default())
    .modify(|deploy| {
        deploy.managed_fields_mut().clear();
        deploy.status = None;
    })
    .applied_objects());

while let Some(d) = truncated_deploy_stream.try_next().await? {
   println!("Truncated Deployment: '{:?}'", serde_json::to_string(&d)?);
}

fn predicate_filter<K, P>( self, predicate: P, config: PredicateConfig, ) -> PredicateFilter<Self, K, P>

where Self: Stream<Item = Result<K, Error>> + Sized, K: Resource + 'static, P: Predicate<K> + 'static,

Filter a stream based on predicates.

This will filter out repeat calls where the predicate returns the same result. Common use case for this is to avoid repeat events for status updates by filtering on predicates::generation.

The cache entries have a configurable time-to-live (TTL) to prevent unbounded memory growth. By default, entries expire after 1 hour.

Deletion handling

This method filters decoded objects without access to their original event type. After touched_objects, a generation-only predicate can suppress objects from delete events because deletion need not change generation. Use predicates::resource_version when observed deletions should remain reconciliation triggers:

let stream = watcher(api, watcher::Config::default())
    .default_backoff()
    .touched_objects()
    .predicate_filter(predicates::resource_version, Default::default());

This also passes status-only changes and does not guarantee delivery of every deletion. See the predicate documentation for choosing a predicate and handling cleanup with finalizers.

Usage

use kube::{Api, Client, ResourceExt};
use kube_runtime::{watcher, WatchStreamExt, predicates};
use k8s_openapi::api::apps::v1::Deployment;
let deploys: Api<Deployment> = Api::default_namespaced(client);
let mut changed_deploys = pin!(watcher(deploys, watcher::Config::default())
    .applied_objects()
    .predicate_filter(predicates::generation, Default::default()));

while let Some(d) = changed_deploys.try_next().await? {
   println!("saw Deployment '{} with hitherto unseen generation", d.name_any());
}

fn reflect<K>(self, writer: Writer<K>) -> Reflect<Self, K>

where Self: Stream<Item = Result<Event<K>>> + Sized, K: Resource + Clone + 'static, K::DynamicType: Eq + Hash + Clone,

Reflect a watcher() stream into a Store through a Writer

Returns the stream unmodified, but passes every watcher::Event through a Writer. This populates a Store as the stream is polled.

Usage

use kube::{Api, Client, ResourceExt};
use kube_runtime::{watcher, WatchStreamExt, reflector};
use k8s_openapi::api::apps::v1::Deployment;

let deploys: Api<Deployment> = Api::default_namespaced(client);
let (reader, writer) = reflector::store::<Deployment>();

tokio::spawn(async move {
    // start polling the store once the reader is ready
    reader.wait_until_ready().await.unwrap();
    loop {
        let names = reader.state().iter().map(|d| d.name_any()).collect::<Vec<_>>();
        info!("Current {} deploys: {:?}", names.len(), names);
        tokio::time::sleep(Duration::from_secs(10)).await;
    }
});

// configure the watcher stream and populate the store while polling
watcher(deploys, watcher::Config::default())
    .reflect(writer)
    .applied_objects()
    .for_each(|res| async move {
        match res {
            Ok(o) => info!("saw {}", o.name_any()),
            Err(e) => warn!("watcher error: {}", e),
        }
    })
    .await;

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl<St> WatchStreamExt for St

where St: Stream + ?Sized,