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

origincloud_core/
operation.rs
