kube_client/api/core_methods.rs
1use either::Either;
2use futures::Stream;
3use serde::{Serialize, de::DeserializeOwned};
4use std::fmt::Debug;
5
6use crate::{Error, Result, api::Api};
7use kube_core::{WatchEvent, metadata::PartialObjectMeta, object::ObjectList, params::*, response::Status};
8
9/// PUSH/PUT/POST/GET abstractions
10impl<K> Api<K>
11where
12 K: Clone + DeserializeOwned + Debug,
13{
14 /// Get a named resource
15 ///
16 /// ```no_run
17 /// # use kube::Api;
18 /// use k8s_openapi::api::core::v1::Pod;
19 ///
20 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
21 /// # let client: kube::Client = todo!();
22 /// let pods: Api<Pod> = Api::namespaced(client, "apps");
23 /// let p: Pod = pods.get("blog").await?;
24 /// # Ok(())
25 /// # }
26 /// ```
27 ///
28 /// # Errors
29 ///
30 /// This function assumes that the object is expected to always exist, and returns [`Error`] if it does not.
31 /// Consider using [`Api::get_opt`] if you need to handle missing objects.
32 pub async fn get(&self, name: &str) -> Result<K> {
33 self.get_with(name, &GetParams::default()).await
34 }
35
36 /// Get only the metadata for a named resource as [`PartialObjectMeta`]
37 ///
38 /// ```no_run
39 /// use kube::{Api, core::PartialObjectMeta};
40 /// use k8s_openapi::api::core::v1::Pod;
41 ///
42 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
43 /// # let client: kube::Client = todo!();
44 /// let pods: Api<Pod> = Api::namespaced(client, "apps");
45 /// let p: PartialObjectMeta<Pod> = pods.get_metadata("blog").await?;
46 /// # Ok(())
47 /// # }
48 /// ```
49 /// Note that the type may be converted to `ObjectMeta` through the usual
50 /// conversion traits.
51 ///
52 /// # Errors
53 ///
54 /// This function assumes that the object is expected to always exist, and returns [`Error`] if it does not.
55 /// Consider using [`Api::get_metadata_opt`] if you need to handle missing objects.
56 pub async fn get_metadata(&self, name: &str) -> Result<PartialObjectMeta<K>> {
57 self.get_metadata_with(name, &GetParams::default()).await
58 }
59
60 /// [Get](`Api::get`) a named resource with an explicit resourceVersion
61 ///
62 /// This function allows the caller to pass in a [`GetParams`](`super::GetParams`) type containing
63 /// a `resourceVersion` to a [Get](`Api::get`) call.
64 /// For example
65 ///
66 /// ```no_run
67 /// # use kube::{Api, api::GetParams};
68 /// use k8s_openapi::api::core::v1::Pod;
69 ///
70 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
71 /// # let client: kube::Client = todo!();
72 /// let pods: Api<Pod> = Api::namespaced(client, "apps");
73 /// let p: Pod = pods.get_with("blog", &GetParams::any()).await?;
74 /// # Ok(())
75 /// # }
76 /// ```
77 ///
78 /// # Errors
79 ///
80 /// This function assumes that the object is expected to always exist, and returns [`Error`] if it does not.
81 /// Consider using [`Api::get_opt`] if you need to handle missing objects.
82 pub async fn get_with(&self, name: &str, gp: &GetParams) -> Result<K> {
83 let mut req = if self.metadata_api {
84 self.request.get_metadata(name, gp)
85 } else {
86 self.request.get(name, gp)
87 }
88 .map_err(Error::BuildRequest)?;
89 req.extensions_mut().insert("get");
90 self.client.request::<K>(req).await
91 }
92
93 /// [Get](`Api::get_metadata`) the metadata of an object using an explicit `resourceVersion`
94 ///
95 /// This function allows the caller to pass in a [`GetParams`](`super::GetParams`) type containing
96 /// a `resourceVersion` to a [Get](`Api::get_metadata`) call.
97 /// For example
98 ///
99 ///
100 /// ```no_run
101 /// use kube::{Api, api::GetParams, core::PartialObjectMeta};
102 /// use k8s_openapi::api::core::v1::Pod;
103 ///
104 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
105 /// # let client: kube::Client = todo!();
106 /// let pods: Api<Pod> = Api::namespaced(client, "apps");
107 /// let p: PartialObjectMeta<Pod> = pods.get_metadata_with("blog", &GetParams::any()).await?;
108 /// # Ok(())
109 /// # }
110 /// ```
111 /// Note that the type may be converted to `ObjectMeta` through the usual
112 /// conversion traits.
113 ///
114 /// # Errors
115 ///
116 /// This function assumes that the object is expected to always exist, and returns [`Error`] if it does not.
117 /// Consider using [`Api::get_metadata_opt`] if you need to handle missing objects.
118 pub async fn get_metadata_with(&self, name: &str, gp: &GetParams) -> Result<PartialObjectMeta<K>> {
119 let mut req = self.request.get_metadata(name, gp).map_err(Error::BuildRequest)?;
120 req.extensions_mut().insert("get_metadata");
121 self.client.request::<PartialObjectMeta<K>>(req).await
122 }
123
124 /// [Get](`Api::get`) a named resource if it exists, returns [`None`] if it doesn't exist
125 ///
126 /// ```no_run
127 /// # use kube::Api;
128 /// use k8s_openapi::api::core::v1::Pod;
129 ///
130 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
131 /// # let client: kube::Client = todo!();
132 /// let pods: Api<Pod> = Api::namespaced(client, "apps");
133 /// if let Some(pod) = pods.get_opt("blog").await? {
134 /// // Pod was found
135 /// } else {
136 /// // Pod was not found
137 /// }
138 /// # Ok(())
139 /// # }
140 /// ```
141 pub async fn get_opt(&self, name: &str) -> Result<Option<K>> {
142 match self.get(name).await {
143 Ok(obj) => Ok(Some(obj)),
144 Err(Error::Api(status)) if status.is_not_found() => Ok(None),
145 Err(err) => Err(err),
146 }
147 }
148
149 /// [Get Metadata](`Api::get_metadata`) for a named resource if it exists, returns [`None`] if it doesn't exist
150 ///
151 /// ```no_run
152 /// # use kube::Api;
153 /// use k8s_openapi::api::core::v1::Pod;
154 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
155 /// # let client: kube::Client = todo!();
156 /// let pods: Api<Pod> = Api::namespaced(client, "apps");
157 /// if let Some(pod) = pods.get_metadata_opt("blog").await? {
158 /// // Pod was found
159 /// } else {
160 /// // Pod was not found
161 /// }
162 /// # Ok(())
163 /// # }
164 /// ```
165 ///
166 /// Note that [`PartialObjectMeta`] embeds the raw `ObjectMeta`.
167 pub async fn get_metadata_opt(&self, name: &str) -> Result<Option<PartialObjectMeta<K>>> {
168 self.get_metadata_opt_with(name, &GetParams::default()).await
169 }
170
171 /// [Get Metadata](`Api::get_metadata`) of an object if it exists, using an explicit `resourceVersion`.
172 /// Returns [`None`] if it doesn't exist.
173 ///
174 /// ```no_run
175 /// # use kube::Api;
176 /// use k8s_openapi::api::core::v1::Pod;
177 /// use kube_core::params::GetParams;
178 ///
179 /// async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
180 /// # let client: kube::Client = todo!();
181 /// let pods: Api<Pod> = Api::namespaced(client, "apps");
182 /// if let Some(pod) = pods.get_metadata_opt_with("blog", &GetParams::any()).await? {
183 /// // Pod was found
184 /// } else {
185 /// // Pod was not found
186 /// }
187 /// # Ok(())
188 /// # }
189 /// ```
190 ///
191 /// Note that [`PartialObjectMeta`] embeds the raw `ObjectMeta`.
192 pub async fn get_metadata_opt_with(
193 &self,
194 name: &str,
195 gp: &GetParams,
196 ) -> Result<Option<PartialObjectMeta<K>>> {
197 match self.get_metadata_with(name, gp).await {
198 Ok(meta) => Ok(Some(meta)),
199 Err(Error::Api(status)) if status.is_not_found() => Ok(None),
200 Err(err) => Err(err),
201 }
202 }
203
204 /// Get a list of resources
205 ///
206 /// You use this to get everything, or a subset matching fields/labels, say:
207 ///
208 /// ```no_run
209 /// use kube::api::{Api, ListParams, ResourceExt};
210 /// use k8s_openapi::api::core::v1::Pod;
211 ///
212 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
213 /// # let client: kube::Client = todo!();
214 /// let pods: Api<Pod> = Api::namespaced(client, "apps");
215 /// let lp = ListParams::default().labels("app=blog"); // for this app only
216 /// for p in pods.list(&lp).await? {
217 /// println!("Found Pod: {}", p.name_any());
218 /// }
219 /// # Ok(())
220 /// # }
221 /// ```
222 pub async fn list(&self, lp: &ListParams) -> Result<ObjectList<K>> {
223 let mut req = if self.metadata_api {
224 self.request.list_metadata(lp)
225 } else {
226 self.request.list(lp)
227 }
228 .map_err(Error::BuildRequest)?;
229 req.extensions_mut().insert("list");
230 self.client.request::<ObjectList<K>>(req).await
231 }
232
233 /// Get a list of resources that contains only their metadata as
234 ///
235 /// Similar to [list](`Api::list`), you use this to get everything, or a
236 /// subset matching fields/labels. For example
237 ///
238 /// ```no_run
239 /// use kube::api::{Api, ListParams, ResourceExt};
240 /// use kube::core::{ObjectMeta, ObjectList, PartialObjectMeta};
241 /// use k8s_openapi::api::core::v1::Pod;
242 ///
243 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
244 /// # let client: kube::Client = todo!();
245 /// let pods: Api<Pod> = Api::namespaced(client, "apps");
246 /// let lp = ListParams::default().labels("app=blog"); // for this app only
247 /// let list: ObjectList<PartialObjectMeta<Pod>> = pods.list_metadata(&lp).await?;
248 /// for p in list {
249 /// println!("Found Pod: {}", p.name_any());
250 /// }
251 /// # Ok(())
252 /// # }
253 /// ```
254 pub async fn list_metadata(&self, lp: &ListParams) -> Result<ObjectList<PartialObjectMeta<K>>> {
255 let mut req = self.request.list_metadata(lp).map_err(Error::BuildRequest)?;
256 req.extensions_mut().insert("list_metadata");
257 self.client.request::<ObjectList<PartialObjectMeta<K>>>(req).await
258 }
259
260 /// Create a resource
261 ///
262 /// This function requires a type that Serializes to `K`, which can be:
263 /// 1. Raw string YAML
264 /// - easy to port from existing files
265 /// - error prone (run-time errors on typos due to failed serialize attempts)
266 /// - very error prone (can write invalid YAML)
267 /// 2. An instance of the struct itself
268 /// - easy to instantiate for CRDs (you define the struct)
269 /// - dense to instantiate for [`k8s_openapi`] types (due to many optionals)
270 /// - compile-time safety
271 /// - but still possible to write invalid native types (validation at apiserver)
272 /// 3. [`serde_json::json!`] macro instantiated [`serde_json::Value`]
273 /// - Tradeoff between the two
274 /// - Easy partially filling of native [`k8s_openapi`] types (most fields optional)
275 /// - Partial safety against runtime errors (at least you must write valid JSON)
276 ///
277 /// Note that this method cannot write to the status object (when it exists) of a resource.
278 /// To set status objects please see [`Api::replace_status`] or [`Api::patch_status`].
279 pub async fn create(&self, pp: &PostParams, data: &K) -> Result<K>
280 where
281 K: Serialize,
282 {
283 let bytes = serde_json::to_vec(&data).map_err(Error::SerdeError)?;
284 let mut req = self.request.create(pp, bytes).map_err(Error::BuildRequest)?;
285 req.extensions_mut().insert("create");
286 self.client.request::<K>(req).await
287 }
288
289 /// Delete a named resource
290 ///
291 /// When you get a `K` via `Left`, your delete has started.
292 /// When you get a `Status` via `Right`, this should be a a 2XX style
293 /// confirmation that the object being gone.
294 ///
295 /// 4XX and 5XX status types are returned as an [`Err(kube_client::Error::Api)`](crate::Error::Api).
296 ///
297 /// ```no_run
298 /// use kube::api::{Api, DeleteParams};
299 /// use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1 as apiexts;
300 /// use apiexts::CustomResourceDefinition;
301 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
302 /// # let client: kube::Client = todo!();
303 /// let crds: Api<CustomResourceDefinition> = Api::all(client);
304 /// crds.delete("foos.clux.dev", &DeleteParams::default()).await?
305 /// .map_left(|o| println!("Deleting CRD: {:?}", o.status))
306 /// .map_right(|s| println!("Deleted CRD: {:?}", s));
307 /// # Ok(())
308 /// # }
309 /// ```
310 pub async fn delete(&self, name: &str, dp: &DeleteParams) -> Result<Either<K, Status>> {
311 let mut req = self.request.delete(name, dp).map_err(Error::BuildRequest)?;
312 req.extensions_mut().insert("delete");
313 self.client.request_status::<K>(req).await
314 }
315
316 /// Delete a collection of resources
317 ///
318 /// When you get an `ObjectList<K>` via `Left`, your delete has started.
319 /// When you get a `Status` via `Right`, this should be a a 2XX style
320 /// confirmation that the object being gone.
321 ///
322 /// 4XX and 5XX status types are returned as an [`Err(kube_client::Error::Api)`](crate::Error::Api).
323 ///
324 /// ```no_run
325 /// use kube::api::{Api, DeleteParams, ListParams, ResourceExt};
326 /// use k8s_openapi::api::core::v1::Pod;
327 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
328 /// # let client: kube::Client = todo!();
329 ///
330 /// let pods: Api<Pod> = Api::namespaced(client, "apps");
331 /// match pods.delete_collection(&DeleteParams::default(), &ListParams::default()).await? {
332 /// either::Left(list) => {
333 /// let names: Vec<_> = list.iter().map(ResourceExt::name_any).collect();
334 /// println!("Deleting collection of pods: {:?}", names);
335 /// },
336 /// either::Right(status) => {
337 /// println!("Deleted collection of pods: status={:?}", status);
338 /// }
339 /// }
340 /// # Ok(())
341 /// # }
342 /// ```
343 pub async fn delete_collection(
344 &self,
345 dp: &DeleteParams,
346 lp: &ListParams,
347 ) -> Result<Either<ObjectList<K>, Status>> {
348 let mut req = self
349 .request
350 .delete_collection(dp, lp)
351 .map_err(Error::BuildRequest)?;
352 req.extensions_mut().insert("delete_collection");
353 self.client.request_status::<ObjectList<K>>(req).await
354 }
355
356 /// Patch a subset of a resource's properties
357 ///
358 /// Takes a [`Patch`] along with [`PatchParams`] for the call.
359 ///
360 /// ```no_run
361 /// use kube::api::{Api, PatchParams, Patch, Resource};
362 /// use k8s_openapi::api::core::v1::Pod;
363 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
364 /// # let client: kube::Client = todo!();
365 ///
366 /// let pods: Api<Pod> = Api::namespaced(client, "apps");
367 /// let patch = serde_json::json!({
368 /// "apiVersion": "v1",
369 /// "kind": "Pod",
370 /// "metadata": {
371 /// "name": "blog"
372 /// },
373 /// "spec": {
374 /// "activeDeadlineSeconds": 5
375 /// }
376 /// });
377 /// let params = PatchParams::apply("myapp");
378 /// let patch = Patch::Apply(&patch);
379 /// let o_patched = pods.patch("blog", ¶ms, &patch).await?;
380 /// # Ok(())
381 /// # }
382 /// ```
383 /// [`Patch`]: super::Patch
384 /// [`PatchParams`]: super::PatchParams
385 ///
386 /// Note that this method cannot write to the status object (when it exists) of a resource.
387 /// To set status objects please see [`Api::replace_status`] or [`Api::patch_status`].
388 pub async fn patch<P: Serialize + Debug>(
389 &self,
390 name: &str,
391 pp: &PatchParams,
392 patch: &Patch<P>,
393 ) -> Result<K> {
394 let mut req = if self.metadata_api {
395 self.request.patch_metadata(name, pp, patch)
396 } else {
397 self.request.patch(name, pp, patch)
398 }
399 .map_err(Error::BuildRequest)?;
400 req.extensions_mut().insert("patch");
401 self.client.request::<K>(req).await
402 }
403
404 /// Patch a metadata subset of a resource's properties from [`PartialObjectMeta`]
405 ///
406 /// Takes a [`Patch`] along with [`PatchParams`] for the call.
407 /// Patches can be constructed raw using `serde_json::json!` or from `ObjectMeta` via [`PartialObjectMetaExt`].
408 ///
409 /// ```no_run
410 /// use kube::api::{Api, PatchParams, Patch, Resource};
411 /// use kube::core::{PartialObjectMetaExt, ObjectMeta};
412 /// use k8s_openapi::api::core::v1::Pod;
413 ///
414 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
415 /// # let client: kube::Client = todo!();
416 /// let pods: Api<Pod> = Api::namespaced(client, "apps");
417 /// let metadata = ObjectMeta {
418 /// labels: Some([("key".to_string(), "value".to_string())].into()),
419 /// ..Default::default()
420 /// }.into_request_partial::<Pod>();
421 ///
422 /// let params = PatchParams::apply("myapp");
423 /// let o_patched = pods.patch_metadata("blog", ¶ms, &Patch::Apply(&metadata)).await?;
424 /// println!("Patched {}", o_patched.metadata.name.unwrap());
425 /// # Ok(())
426 /// # }
427 /// ```
428 /// [`Patch`]: super::Patch
429 /// [`PatchParams`]: super::PatchParams
430 /// [`PartialObjectMetaExt`]: crate::core::PartialObjectMetaExt
431 ///
432 /// ### Warnings
433 ///
434 /// The `TypeMeta` (apiVersion + kind) of a patch request (required for apply patches)
435 /// must match the underlying type that is being patched (e.g. "v1" + "Pod").
436 /// The returned `TypeMeta` will always be {"meta.k8s.io/v1", "PartialObjectMetadata"}.
437 /// These constraints are encoded into [`PartialObjectMetaExt`].
438 ///
439 /// This method can write to non-metadata fields such as spec if included in the patch.
440 pub async fn patch_metadata<P: Serialize + Debug>(
441 &self,
442 name: &str,
443 pp: &PatchParams,
444 patch: &Patch<P>,
445 ) -> Result<PartialObjectMeta<K>> {
446 let mut req = self
447 .request
448 .patch_metadata(name, pp, patch)
449 .map_err(Error::BuildRequest)?;
450 req.extensions_mut().insert("patch_metadata");
451 self.client.request::<PartialObjectMeta<K>>(req).await
452 }
453
454 /// Replace a resource entirely with a new one
455 ///
456 /// This is used just like [`Api::create`], but with one additional instruction:
457 /// You must set `metadata.resourceVersion` in the provided data because k8s
458 /// will not accept an update unless you actually knew what the last version was.
459 ///
460 /// Thus, to use this function, you need to do a `get` then a `replace` with its result.
461 ///
462 /// ```no_run
463 /// use kube::api::{Api, PostParams, ResourceExt};
464 /// use k8s_openapi::api::batch::v1::Job;
465 ///
466 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
467 /// # let client: kube::Client = todo!();
468 /// let jobs: Api<Job> = Api::namespaced(client, "apps");
469 /// let j = jobs.get("baz").await?;
470 /// let j_new: Job = serde_json::from_value(serde_json::json!({
471 /// "apiVersion": "batch/v1",
472 /// "kind": "Job",
473 /// "metadata": {
474 /// "name": "baz",
475 /// "resourceVersion": j.resource_version(),
476 /// },
477 /// "spec": {
478 /// "template": {
479 /// "metadata": {
480 /// "name": "empty-job-pod"
481 /// },
482 /// "spec": {
483 /// "containers": [{
484 /// "name": "empty",
485 /// "image": "alpine:latest"
486 /// }],
487 /// "restartPolicy": "Never",
488 /// }
489 /// }
490 /// }
491 /// }))?;
492 /// jobs.replace("baz", &PostParams::default(), &j_new).await?;
493 /// # Ok(())
494 /// # }
495 /// ```
496 ///
497 /// Consider mutating the result of `api.get` rather than recreating it.
498 ///
499 /// Note that this method cannot write to the status object (when it exists) of a resource.
500 /// To set status objects please see [`Api::replace_status`] or [`Api::patch_status`].
501 pub async fn replace(&self, name: &str, pp: &PostParams, data: &K) -> Result<K>
502 where
503 K: Serialize,
504 {
505 let bytes = serde_json::to_vec(&data).map_err(Error::SerdeError)?;
506 let mut req = self
507 .request
508 .replace(name, pp, bytes)
509 .map_err(Error::BuildRequest)?;
510 req.extensions_mut().insert("replace");
511 self.client.request::<K>(req).await
512 }
513
514 /// Watch a list of resources
515 ///
516 /// This returns a future that awaits the initial response,
517 /// then you can stream the remaining buffered `WatchEvent` objects.
518 ///
519 /// Note that a `watch` call can terminate for many reasons (even before the specified
520 /// [`WatchParams::timeout`] is triggered), and will have to be re-issued
521 /// with the last seen resource version when or if it closes.
522 ///
523 /// Consider using a managed [`watcher`] to deal with automatic re-watches and error cases.
524 ///
525 /// ```no_run
526 /// use kube::api::{Api, WatchParams, ResourceExt, WatchEvent};
527 /// use k8s_openapi::api::batch::v1::Job;
528 /// use futures::{StreamExt, TryStreamExt};
529 ///
530 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
531 /// # let client: kube::Client = todo!();
532 /// let jobs: Api<Job> = Api::namespaced(client, "apps");
533 /// let lp = WatchParams::default()
534 /// .fields("metadata.name=my_job")
535 /// .timeout(20); // upper bound of how long we watch for
536 /// let mut stream = jobs.watch(&lp, "0").await?.boxed();
537 /// while let Some(status) = stream.try_next().await? {
538 /// match status {
539 /// WatchEvent::Added(s) => println!("Added {}", s.name_any()),
540 /// WatchEvent::Modified(s) => println!("Modified: {}", s.name_any()),
541 /// WatchEvent::Deleted(s) => println!("Deleted {}", s.name_any()),
542 /// WatchEvent::Bookmark(s) => {},
543 /// WatchEvent::Error(s) => println!("{}", s),
544 /// }
545 /// }
546 /// # Ok(())
547 /// # }
548 /// ```
549 /// [`WatchParams::timeout`]: super::WatchParams::timeout
550 /// [`watcher`]: https://docs.rs/kube_runtime/*/kube_runtime/watcher/fn.watcher.html
551 pub async fn watch(
552 &self,
553 wp: &WatchParams,
554 version: &str,
555 ) -> Result<impl Stream<Item = Result<WatchEvent<K>>> + use<K>> {
556 let mut req = if self.metadata_api {
557 self.request.watch_metadata(wp, version)
558 } else {
559 self.request.watch(wp, version)
560 }
561 .map_err(Error::BuildRequest)?;
562 req.extensions_mut().insert("watch");
563 self.client.request_events::<K>(req).await
564 }
565
566 /// Watch a list of metadata for a given resources
567 ///
568 /// This returns a future that awaits the initial response,
569 /// then you can stream the remaining buffered `WatchEvent` objects.
570 ///
571 /// Note that a `watch_metadata` call can terminate for many reasons (even
572 /// before the specified [`WatchParams::timeout`] is triggered), and will
573 /// have to be re-issued with the last seen resource version when or if it
574 /// closes.
575 ///
576 /// Consider using a managed [`metadata_watcher`] to deal with automatic re-watches and error cases.
577 ///
578 /// ```no_run
579 /// use kube::api::{Api, WatchParams, ResourceExt, WatchEvent};
580 /// use k8s_openapi::api::batch::v1::Job;
581 /// use futures::{StreamExt, TryStreamExt};
582 ///
583 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
584 /// # let client: kube::Client = todo!();
585 /// let jobs: Api<Job> = Api::namespaced(client, "apps");
586 ///
587 /// let lp = WatchParams::default()
588 /// .fields("metadata.name=my_job")
589 /// .timeout(20); // upper bound of how long we watch for
590 /// let mut stream = jobs.watch(&lp, "0").await?.boxed();
591 /// while let Some(status) = stream.try_next().await? {
592 /// match status {
593 /// WatchEvent::Added(s) => println!("Added {}", s.metadata.name.unwrap()),
594 /// WatchEvent::Modified(s) => println!("Modified: {}", s.metadata.name.unwrap()),
595 /// WatchEvent::Deleted(s) => println!("Deleted {}", s.metadata.name.unwrap()),
596 /// WatchEvent::Bookmark(s) => {},
597 /// WatchEvent::Error(s) => println!("{}", s),
598 /// }
599 /// }
600 /// # Ok(())
601 /// # }
602 /// ```
603 /// [`WatchParams::timeout`]: super::WatchParams::timeout
604 /// [`metadata_watcher`]: https://docs.rs/kube_runtime/*/kube_runtime/watcher/fn.metadata_watcher.html
605 pub async fn watch_metadata(
606 &self,
607 wp: &WatchParams,
608 version: &str,
609 ) -> Result<impl Stream<Item = Result<WatchEvent<PartialObjectMeta<K>>>> + use<K>> {
610 let mut req = self
611 .request
612 .watch_metadata(wp, version)
613 .map_err(Error::BuildRequest)?;
614 req.extensions_mut().insert("watch_metadata");
615 self.client.request_events::<PartialObjectMeta<K>>(req).await
616 }
617}