origincloud_core/
operation.rs

1use std::fmt;
2use std::marker::PhantomData;
3use std::time::Duration;
4
5use serde::de::DeserializeOwned;
6
7use crate::error::ApiError;
8use crate::longrunning::OperationResult;
9use crate::transport::is_timeout;
10use crate::{Error, Method, Request, Service, longrunning, path_value};
11
12/// A mutation the API accepted and carries out after answering. Its
13/// result, a `T`, is the resource it acted on, or
14/// [`Empty`](crate::protobuf::Empty) for a delete.
15pub struct Operation<T> {
16    service: Service,
17    operation: longrunning::Operation,
18    result_type: &'static str,
19    result: PhantomData<fn() -> T>,
20}
21
22impl<T: DeserializeOwned> Operation<T> {
23    /// The operation `operation`, which a method of `service` answered
24    /// with; its result is the message `result_type`, such as
25    /// `origincloud.pvs.v1beta1.Volume`. The generated clients call it.
26    /// Waiting on it follows the client's settings, not the
27    /// [`CallOptions`](crate::CallOptions) of `service`.
28    pub fn new(
29        service: Service,
30        operation: longrunning::Operation,
31        result_type: &'static str,
32    ) -> Self {
33        Self {
34            service: service.with_options(crate::CallOptions::default()),
35            operation,
36            result_type,
37            result: PhantomData,
38        }
39    }
40
41    /// The operation's resource name, by which [`Operation::wait`] and a
42    /// service's `OperationClient` find it.
43    pub fn name(&self) -> &str {
44        &self.operation.name
45    }
46
47    /// Whether the operation had finished when it was last read.
48    pub fn done(&self) -> bool {
49        self.operation.done
50    }
51
52    /// The operation as it was last read, its metadata included.
53    pub fn raw(&self) -> &longrunning::Operation {
54        &self.operation
55    }
56
57    /// Waits until the operation finishes and returns its result, or the
58    /// [`Error::Api`] it failed with. Once it returns a result, reads
59    /// through the same [`Client`](crate::Client) show the operation's
60    /// effect.
61    ///
62    /// Each call asks the API to wait up to
63    /// [`ClientBuilder::operation_wait`](crate::ClientBuilder::operation_wait),
64    /// a minute by default, and less than the client's
65    /// [`timeout`](crate::ClientBuilder::timeout). A call that times out
66    /// all the same, such as under a shorter timeout of the HTTP client's
67    /// own, counts as the operation not being done yet: the next call is
68    /// sent, without using up the call's retries.
69    ///
70    /// It waits as long as the operation takes; to bound that, wrap it in a
71    /// timeout such as `tokio::time::timeout`, or in a browser a race
72    /// against a timer (`futures::future::select` with
73    /// `gloo_timers::future::sleep`), which stops waiting, not the
74    /// operation.
75    pub async fn wait(&mut self) -> Result<T, Error> {
76        let body = serde_json::json!({ "timeout": format!("{}s", self.wait_per_call().as_secs()) });
77        while !self.operation.done {
78            let path = path_value(
79                "operation name",
80                &self.operation.name,
81                "projects/*/zones/*/operations/*",
82            )?;
83            let mut call = Request::new(Method::POST, format!("/v1beta1/{path}:wait"))
84                .with_body(body.to_string().into_bytes())
85                .idempotent();
86            call.timeout_answers = true;
87            match self.service.call(call).await {
88                Ok(operation) => self.operation = operation,
89                Err(err) if is_timeout(&err) => {}
90                Err(err) => return Err(err),
91            }
92        }
93        self.result()
94    }
95
96    /// How long one wait call asks the API to wait: the client's
97    /// `operation_wait`, kept a second, or half, below its timeout, in
98    /// whole seconds, at least one.
99    fn wait_per_call(&self) -> Duration {
100        let inner = &self.service.client.inner;
101        let mut wait = inner.operation_wait;
102        if let Some(timeout) = inner.timeout {
103            let margin = Duration::from_secs(1).min(timeout / 2);
104            wait = wait.min(timeout - margin);
105        }
106        Duration::from_secs(wait.as_secs().max(1))
107    }
108
109    /// The result of an operation that is done, or the [`Error::Api`] it
110    /// failed with. It fails with [`Error::NotDone`] for an operation that
111    /// isn't done yet.
112    pub fn result(&self) -> Result<T, Error> {
113        let name = &self.operation.name;
114        if !self.operation.done {
115            return Err(Error::NotDone(name.clone()));
116        }
117        match &self.operation.result {
118            Some(OperationResult::Error(status)) => Err(ApiError::from_status(status).into()),
119            Some(OperationResult::Response(response)) => {
120                if response.type_name() != self.result_type {
121                    return Err(Error::Decode(format!(
122                        "the result of operation {name} is a {}, not a {}",
123                        response.type_name(),
124                        self.result_type
125                    )));
126                }
127                response.decode().map_err(|e| {
128                    Error::Decode(format!("reading the result of operation {name}: {e}"))
129                })
130            }
131            None => Err(Error::Decode(format!(
132                "operation {name} is done without a result"
133            ))),
134        }
135    }
136}
137
138impl<T> fmt::Debug for Operation<T> {
139    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
140        f.debug_struct("Operation")
141            .field("name", &self.operation.name)
142            .field("done", &self.operation.done)
143            .field("result_type", &self.result_type)
144            .finish_non_exhaustive()
145    }
146}
origincloud_core/operation.rs

origincloud_core/
operation.rs
