temporalio_client/activity/
activity_handle.rs

1use crate::{
2    ActivityCancelOptions, ActivityDescribeOptions, ActivityExecutionDescription,
3    ActivityTerminateOptions, NamespacedClient,
4    errors::{ActivityInteractionError, ActivityResultError},
5    grpc::WorkflowService,
6};
7use std::marker::PhantomData;
8use temporalio_common::{
9    ActivityDefinition,
10    data_converters::{DecodablePayloads, NoopDecodeHint, SerializationContextData},
11    protos::temporal::api::{
12        activity::v1::{ActivityExecutionOutcome, activity_execution_outcome},
13        failure::v1::failure::FailureInfo,
14        workflowservice::v1::{
15            DescribeActivityExecutionRequest, PollActivityExecutionRequest,
16            RequestCancelActivityExecutionRequest, TerminateActivityExecutionRequest,
17        },
18    },
19};
20use tonic::IntoRequest;
21use uuid::Uuid;
22
23/// Handle associated with a standalone activity execution that can be used to wait for the result
24/// or to manage execution of the activity. Obtained from
25/// [`Client::start_activity`](crate::Client::start_activity) or
26/// [`Client::get_activity_handle`](crate::Client::get_activity_handle).
27///
28/// If [`run_id`](Self::run_id) is set, the handle always targets that specific execution.
29/// If [`run_id`](Self::run_id) is `None`, each method call targets the latest run of the specified
30/// [`activity_id`](Self::activity_id) at the time the method is called - this means consecutive
31/// method calls may target different executions if an activity was started again with the same ID.
32pub struct ActivityHandle<ClientT, ActivityT>
33where
34    ActivityT: ActivityDefinition,
35{
36    client: ClientT,
37    activity_id: String,
38    run_id: Option<String>,
39    _phantom: PhantomData<ActivityT>,
40}
41
42impl<ClientT, ActivityT> ActivityHandle<ClientT, ActivityT>
43where
44    ActivityT: ActivityDefinition,
45{
46    pub(crate) fn new(client: ClientT, activity_id: String, run_id: Option<String>) -> Self {
47        Self {
48            client,
49            activity_id,
50            run_id,
51            _phantom: PhantomData,
52        }
53    }
54
55    /// Activity ID this handle is associated with.
56    pub fn activity_id(&self) -> &str {
57        &self.activity_id
58    }
59
60    /// Run ID of the activity execution this handle is associated with. If `None`, each method call
61    /// targets the latest run of the specified [`activity_id`](Self::activity_id) at the time the
62    /// method is called - this means consecutive method calls may target different executions if
63    /// an activity was started again with the same ID.
64    pub fn run_id(&self) -> Option<&str> {
65        self.run_id.as_deref()
66    }
67}
68
69impl<ClientT, ActivityT> ActivityHandle<ClientT, ActivityT>
70where
71    ClientT: WorkflowService + NamespacedClient + Clone,
72    ActivityT: ActivityDefinition,
73{
74    /// Wait for the activity to complete and fetch its result. If the activity was not successful
75    /// (e.g. failed, canceled, timed out), this method returns [`ActivityResultError::ActivityFailed`].
76    pub async fn result(&self) -> Result<ActivityT::Output, ActivityResultError> {
77        let mut client = self.client.clone();
78        loop {
79            let resp = client
80                .poll_activity_execution(
81                    PollActivityExecutionRequest {
82                        namespace: client.namespace(),
83                        activity_id: self.activity_id.clone(),
84                        run_id: self.run_id.clone().unwrap_or_default(),
85                    }
86                    .into_request(),
87                )
88                .await?
89                .into_inner();
90
91            // If resp.outcome.value is None, poll again
92            let Some(ActivityExecutionOutcome {
93                value: Some(outcome),
94                ..
95            }) = resp.outcome
96            else {
97                continue;
98            };
99
100            let dc = client.data_converter();
101            let ctx = SerializationContextData::Activity;
102
103            return match outcome {
104                activity_execution_outcome::Value::Result(payloads) => {
105                    Ok(dc.from_payloads(&ctx, payloads.payloads).await?)
106                }
107                activity_execution_outcome::Value::Failure(failure) => {
108                    Err(match failure.failure_info {
109                        Some(FailureInfo::CanceledFailureInfo(info)) => {
110                            let payloads = info.details.unwrap_or_default().payloads;
111                            let details = DecodablePayloads::new(
112                                payloads,
113                                dc.payload_converter().clone(),
114                                ctx,
115                            );
116                            ActivityResultError::Cancelled { details }
117                        }
118                        Some(FailureInfo::TerminatedFailureInfo(_)) => {
119                            ActivityResultError::Terminated
120                        }
121                        _ => ActivityResultError::ActivityFailed(dc.to_error(
122                            &ctx,
123                            failure,
124                            NoopDecodeHint,
125                        )?),
126                    })
127                }
128            };
129        }
130    }
131
132    /// Describes the current state of the activity execution.
133    pub async fn describe(
134        &self,
135        options: ActivityDescribeOptions,
136    ) -> Result<ActivityExecutionDescription<ActivityT>, ActivityInteractionError> {
137        let mut client = self.client.clone();
138        let resp = client
139            .describe_activity_execution(
140                DescribeActivityExecutionRequest {
141                    namespace: client.namespace(),
142                    activity_id: self.activity_id.clone(),
143                    run_id: self.run_id.clone().unwrap_or_default(),
144                    include_input: options.include_input,
145                    include_outcome: options.include_outcome,
146                    include_heartbeat_details: options.include_heartbeat_details,
147                    include_last_failure: options.include_last_failure,
148                    ..Default::default()
149                }
150                .into_request(),
151            )
152            .await?
153            .into_inner();
154
155        Ok(ActivityExecutionDescription::new(
156            client.data_converter().clone(),
157            SerializationContextData::Activity,
158            resp,
159        )?)
160    }
161
162    /// Requests cancellation of the activity. Does not wait for the cancellation to complete.
163    pub async fn cancel(
164        &self,
165        options: ActivityCancelOptions,
166    ) -> Result<(), ActivityInteractionError> {
167        let mut client = self.client.clone();
168        client
169            .request_cancel_activity_execution(
170                RequestCancelActivityExecutionRequest {
171                    namespace: client.namespace(),
172                    activity_id: self.activity_id.clone(),
173                    run_id: self.run_id.clone().unwrap_or_default(),
174                    identity: client.identity(),
175                    request_id: Uuid::new_v4().to_string(),
176                    reason: options.reason,
177                }
178                .into_request(),
179            )
180            .await?;
181
182        Ok(())
183    }
184
185    /// Terminates activity execution.
186    pub async fn terminate(
187        &self,
188        options: ActivityTerminateOptions,
189    ) -> Result<(), ActivityInteractionError> {
190        let mut client = self.client.clone();
191        client
192            .terminate_activity_execution(
193                TerminateActivityExecutionRequest {
194                    namespace: client.namespace(),
195                    activity_id: self.activity_id.clone(),
196                    run_id: self.run_id.clone().unwrap_or_default(),
197                    identity: client.identity(),
198                    request_id: Uuid::new_v4().to_string(),
199                    reason: options.reason,
200                }
201                .into_request(),
202            )
203            .await?;
204
205        Ok(())
206    }
207}
temporalio_client/activity/activity_handle.rs

temporalio_client/activity/
activity_handle.rs
