temporalio_client/
async_activity_handle.rs

1//! Handle for completing activities asynchronously via a client.
2
3use crate::{
4    CompleteAsyncActivityInput, FailAsyncActivityInput, HeartbeatAsyncActivityInput,
5    NamespacedClient, Next, ReportAsyncActivityCancellationInput, RpcOptions, TemporalClientValue,
6    errors::AsyncActivityError, grpc::WorkflowService, interceptors,
7};
8use futures_util::future::BoxFuture;
9use temporalio_common::{
10    data_converters::{SerializationContext, SerializationContextData, TemporalSerializable},
11    error::{ApplicationFailure, OutgoingActivityError, OutgoingError},
12    payload_visitor::encode_payloads,
13    protos::{
14        TaskToken,
15        temporal::api::{
16            common::v1::Payloads,
17            workflowservice::v1::{
18                RecordActivityTaskHeartbeatByIdRequest, RecordActivityTaskHeartbeatByIdResponse,
19                RecordActivityTaskHeartbeatRequest, RecordActivityTaskHeartbeatResponse,
20                RespondActivityTaskCanceledByIdRequest, RespondActivityTaskCanceledRequest,
21                RespondActivityTaskCompletedByIdRequest, RespondActivityTaskCompletedRequest,
22                RespondActivityTaskFailedByIdRequest, RespondActivityTaskFailedRequest,
23            },
24        },
25    },
26};
27use tonic::IntoRequest;
28
29async fn encode_optional_value(
30    value: Option<Box<dyn TemporalClientValue>>,
31    data_converter: &temporalio_common::data_converters::DataConverter,
32) -> Result<Option<Payloads>, AsyncActivityError> {
33    let Some(value) = value else {
34        return Ok(None);
35    };
36    let unencoded_payloads = {
37        let payload_converter = data_converter.payload_converter();
38        let context = SerializationContext {
39            data: &SerializationContextData::Activity,
40            converter: payload_converter,
41        };
42        value.serialize_payloads(&context)?
43    };
44    drop(value);
45    let payloads = data_converter
46        .codec()
47        .encode(&SerializationContextData::Activity, unencoded_payloads)
48        .await?;
49    Ok(Some(Payloads { payloads }))
50}
51
52/// Identifies an async activity for completion outside a worker.
53#[derive(Debug, Clone)]
54pub enum ActivityIdentifier {
55    /// Identify activity by its task token
56    TaskToken(TaskToken),
57    /// Identify workflow activity by workflow and activity IDs.
58    ByIdWorkflow {
59        /// ID of the workflow that scheduled this activity.
60        workflow_id: String,
61        /// Run ID of the workflow (optional - if not provided, targets the latest run).
62        run_id: String,
63        /// ID of the activity to complete.
64        activity_id: String,
65    },
66    /// Identify standalone activity by activity ID.
67    ByIdStandalone {
68        /// ID of the activity to complete.
69        activity_id: String,
70        /// Run ID of the activity (optional - if not provided, targets the latest run).
71        run_id: String,
72    },
73}
74
75impl ActivityIdentifier {
76    /// Create an identifier from a task token.
77    pub fn from_task_token(token: TaskToken) -> Self {
78        Self::TaskToken(token)
79    }
80
81    /// Create an identifier from workflow and activity IDs. Use an empty run id to target the
82    /// latest workflow execution.
83    pub fn by_id_workflow(
84        workflow_id: impl Into<String>,
85        run_id: impl Into<String>,
86        activity_id: impl Into<String>,
87    ) -> Self {
88        Self::ByIdWorkflow {
89            workflow_id: workflow_id.into(),
90            run_id: run_id.into(),
91            activity_id: activity_id.into(),
92        }
93    }
94
95    /// Create an identifier from standalone activity ID. Use an empty run id to target the
96    /// latest activity execution.
97    pub fn by_id_standalone(activity_id: impl Into<String>, run_id: impl Into<String>) -> Self {
98        Self::ByIdStandalone {
99            activity_id: activity_id.into(),
100            run_id: run_id.into(),
101        }
102    }
103
104    /// Returns tuple of (workflow_id, run_id, activity_id).
105    fn into_parts(self) -> Option<(String, String, String)> {
106        match self {
107            Self::TaskToken(_) => None,
108            Self::ByIdWorkflow {
109                workflow_id,
110                run_id,
111                activity_id,
112            } => Some((workflow_id, run_id, activity_id)),
113            Self::ByIdStandalone {
114                activity_id,
115                run_id,
116            } => Some((String::new(), run_id, activity_id)),
117        }
118    }
119}
120
121/// Handle for completing activities asynchronously (outside the worker).
122pub struct AsyncActivityHandle<CT> {
123    client: CT,
124    identifier: ActivityIdentifier,
125}
126
127impl<CT> AsyncActivityHandle<CT> {
128    /// Create a new async activity handle.
129    pub fn new(client: CT, identifier: ActivityIdentifier) -> Self {
130        Self { client, identifier }
131    }
132
133    /// Get the identifier for this activity.
134    pub fn identifier(&self) -> &ActivityIdentifier {
135        &self.identifier
136    }
137
138    /// Get a reference to the underlying client.
139    pub fn client(&self) -> &CT {
140        &self.client
141    }
142}
143
144impl<CT: WorkflowService + NamespacedClient + Clone> AsyncActivityHandle<CT> {
145    /// Complete the activity with a successful result.
146    pub async fn complete<T>(
147        &self,
148        result: Option<T>,
149        rpc_options: RpcOptions,
150    ) -> Result<(), AsyncActivityError>
151    where
152        T: TemporalSerializable + Send + 'static,
153    {
154        interceptors::call_complete_async_activity(
155            self.client.client_interceptors(),
156            CompleteAsyncActivityInput::new(self.identifier.clone(), result, rpc_options),
157            Next::new({
158                let mut client = self.client.clone();
159                move |input: CompleteAsyncActivityInput| -> BoxFuture<
160                    '_,
161                    Result<(), AsyncActivityError>,
162                > {
163                    Box::pin(async move {
164                        let (identifier, result, rpc_options) = input.into_parts();
165                        let result = encode_optional_value(result, client.data_converter()).await?;
166                        if let ActivityIdentifier::TaskToken(token) = identifier {
167                            let mut request = RespondActivityTaskCompletedRequest {
168                                task_token: token.0,
169                                result,
170                                identity: client.identity(),
171                                namespace: client.namespace(),
172                                ..Default::default()
173                            }
174                            .into_request();
175                            rpc_options.apply_to(&mut request);
176                            WorkflowService::respond_activity_task_completed(
177                                &mut client,
178                                request,
179                            )
180                            .await
181                            .map_err(AsyncActivityError::from_status)?;
182                        } else {
183                            let (workflow_id, run_id, activity_id) = identifier.into_parts().unwrap();
184                            let mut request = RespondActivityTaskCompletedByIdRequest {
185                                namespace: client.namespace(),
186                                workflow_id,
187                                run_id,
188                                activity_id,
189                                result,
190                                identity: client.identity(),
191                                resource_id: Default::default(),
192                            }
193                            .into_request();
194                            rpc_options.apply_to(&mut request);
195                            WorkflowService::respond_activity_task_completed_by_id(
196                                &mut client,
197                                request,
198                            )
199                            .await
200                            .map_err(AsyncActivityError::from_status)?;
201                        }
202                        Ok(())
203                    })
204                }
205            }),
206        )
207        .await
208    }
209
210    /// Fail the activity with a failure.
211    pub async fn fail<E, T>(
212        &self,
213        failure: E,
214        last_heartbeat_details: Option<T>,
215        rpc_options: RpcOptions,
216    ) -> Result<(), AsyncActivityError>
217    where
218        E: Into<ApplicationFailure>,
219        T: TemporalSerializable + Send + 'static,
220    {
221        interceptors::call_fail_async_activity(
222            self.client.client_interceptors(),
223            FailAsyncActivityInput::new(
224                self.identifier.clone(),
225                failure.into(),
226                last_heartbeat_details,
227                rpc_options,
228            ),
229            Next::new({
230                let mut client = self.client.clone();
231                move |input: FailAsyncActivityInput| -> BoxFuture<
232                    '_,
233                    Result<(), AsyncActivityError>,
234                > {
235                    Box::pin(async move {
236                        let (identifier, application_failure, details, rpc_options) =
237                            input.into_parts();
238                        let data_converter = client.data_converter().clone();
239                        let mut failure = data_converter.to_failure(
240                            &SerializationContextData::Activity,
241                            OutgoingError::Activity(OutgoingActivityError::Application(Box::new(
242                                application_failure,
243                            ))),
244                        );
245                        encode_payloads(
246                            &mut failure,
247                            data_converter.codec(),
248                            &SerializationContextData::Activity,
249                        )
250                        .await?;
251                        let last_heartbeat_details =
252                            encode_optional_value(details, &data_converter).await?;
253                        if let ActivityIdentifier::TaskToken(token) = identifier {
254                            let mut request = RespondActivityTaskFailedRequest {
255                                task_token: token.0,
256                                failure: Some(failure),
257                                identity: client.identity(),
258                                namespace: client.namespace(),
259                                last_heartbeat_details,
260                                ..Default::default()
261                            }
262                            .into_request();
263                            rpc_options.apply_to(&mut request);
264                            WorkflowService::respond_activity_task_failed(
265                                &mut client,
266                                request,
267                            )
268                            .await
269                            .map_err(AsyncActivityError::from_status)?;
270                        } else {
271                            let (workflow_id, run_id, activity_id) = identifier.into_parts().unwrap();
272                            let mut request = RespondActivityTaskFailedByIdRequest {
273                                namespace: client.namespace(),
274                                workflow_id,
275                                run_id,
276                                activity_id,
277                                failure: Some(failure),
278                                identity: client.identity(),
279                                last_heartbeat_details,
280                                resource_id: Default::default(),
281                            }
282                            .into_request();
283                            rpc_options.apply_to(&mut request);
284                            WorkflowService::respond_activity_task_failed_by_id(
285                                &mut client,
286                                request,
287                            )
288                            .await
289                            .map_err(AsyncActivityError::from_status)?;
290                        }
291                        Ok(())
292                    })
293                }
294            }),
295        )
296        .await
297    }
298
299    /// Reports the activity as canceled.
300    pub async fn report_cancelation<T>(
301        &self,
302        details: Option<T>,
303        rpc_options: RpcOptions,
304    ) -> Result<(), AsyncActivityError>
305    where
306        T: TemporalSerializable + Send + 'static,
307    {
308        interceptors::call_report_async_activity_cancellation(
309            self.client.client_interceptors(),
310            ReportAsyncActivityCancellationInput::new(
311                self.identifier.clone(),
312                details,
313                rpc_options,
314            ),
315            Next::new({
316                let mut client = self.client.clone();
317                move |input: ReportAsyncActivityCancellationInput| -> BoxFuture<
318                    '_,
319                    Result<(), AsyncActivityError>,
320                > {
321                    Box::pin(async move {
322                        let (identifier, details, rpc_options) = input.into_parts();
323                        let details = encode_optional_value(details, client.data_converter()).await?;
324                        if let ActivityIdentifier::TaskToken(token) = identifier {
325                            let mut request = RespondActivityTaskCanceledRequest {
326                                task_token: token.0,
327                                details,
328                                identity: client.identity(),
329                                namespace: client.namespace(),
330                                ..Default::default()
331                            }
332                            .into_request();
333                            rpc_options.apply_to(&mut request);
334                            WorkflowService::respond_activity_task_canceled(
335                                &mut client,
336                                request,
337                            )
338                            .await
339                            .map_err(AsyncActivityError::from_status)?;
340                        } else {
341                            let (workflow_id, run_id, activity_id) = identifier.into_parts().unwrap();
342                            let mut request = RespondActivityTaskCanceledByIdRequest {
343                                namespace: client.namespace(),
344                                workflow_id,
345                                run_id,
346                                activity_id,
347                                details,
348                                identity: client.identity(),
349                                ..Default::default()
350                            }
351                            .into_request();
352                            rpc_options.apply_to(&mut request);
353                            WorkflowService::respond_activity_task_canceled_by_id(
354                                &mut client,
355                                request,
356                            )
357                            .await
358                            .map_err(AsyncActivityError::from_status)?;
359                        }
360                        Ok(())
361                    })
362                }
363            }),
364        )
365        .await
366    }
367
368    /// Record a heartbeat for the activity.
369    ///
370    /// Heartbeats let the server know the activity is still running and can carry
371    /// progress information. The response indicates if cancellation has been requested.
372    pub async fn heartbeat<T>(
373        &self,
374        details: Option<T>,
375        rpc_options: RpcOptions,
376    ) -> Result<ActivityHeartbeatResponse, AsyncActivityError>
377    where
378        T: TemporalSerializable + Send + 'static,
379    {
380        interceptors::call_heartbeat_async_activity(
381            self.client.client_interceptors(),
382            HeartbeatAsyncActivityInput::new(self.identifier.clone(), details, rpc_options),
383            Next::new({
384                let mut client = self.client.clone();
385                move |input: HeartbeatAsyncActivityInput| -> BoxFuture<
386                    '_,
387                    Result<ActivityHeartbeatResponse, AsyncActivityError>,
388                > {
389                    Box::pin(async move {
390                        let (identifier, details, rpc_options) = input.into_parts();
391                        let details = encode_optional_value(details, client.data_converter()).await?;
392                        if let ActivityIdentifier::TaskToken(token) = identifier {
393                            let mut request = RecordActivityTaskHeartbeatRequest {
394                                task_token: token.0,
395                                details,
396                                identity: client.identity(),
397                                namespace: client.namespace(),
398                                resource_id: Default::default(),
399                            }
400                            .into_request();
401                            rpc_options.apply_to(&mut request);
402                            let response = WorkflowService::record_activity_task_heartbeat(
403                                &mut client,
404                                request,
405                            )
406                            .await
407                            .map_err(AsyncActivityError::from_status)?
408                            .into_inner();
409                            Ok(ActivityHeartbeatResponse::from(response))
410                        } else {
411                            let (workflow_id, run_id, activity_id) = identifier.into_parts().unwrap();
412                            let mut request = RecordActivityTaskHeartbeatByIdRequest {
413                                namespace: client.namespace(),
414                                workflow_id,
415                                run_id,
416                                activity_id,
417                                details,
418                                identity: client.identity(),
419                                resource_id: Default::default(),
420                            }
421                            .into_request();
422                            rpc_options.apply_to(&mut request);
423                            let response =
424                                WorkflowService::record_activity_task_heartbeat_by_id(
425                                    &mut client,
426                                    request,
427                                )
428                                .await
429                                .map_err(AsyncActivityError::from_status)?
430                                .into_inner();
431                            Ok(ActivityHeartbeatResponse::from(response))
432                        }
433                    })
434                }
435            }),
436        )
437        .await
438    }
439}
440
441/// Response from a heartbeat call.
442#[derive(Debug, Clone)]
443pub struct ActivityHeartbeatResponse {
444    /// True if the activity has been asked to cancel itself.
445    pub cancel_requested: bool,
446    /// True if the activity is paused.
447    pub activity_paused: bool,
448    /// True if the activity was reset.
449    pub activity_reset: bool,
450}
451
452impl From<RecordActivityTaskHeartbeatResponse> for ActivityHeartbeatResponse {
453    fn from(resp: RecordActivityTaskHeartbeatResponse) -> Self {
454        Self {
455            cancel_requested: resp.cancel_requested,
456            activity_paused: resp.activity_paused,
457            activity_reset: resp.activity_reset,
458        }
459    }
460}
461
462impl From<RecordActivityTaskHeartbeatByIdResponse> for ActivityHeartbeatResponse {
463    fn from(resp: RecordActivityTaskHeartbeatByIdResponse) -> Self {
464        Self {
465            cancel_requested: resp.cancel_requested,
466            activity_paused: resp.activity_paused,
467            activity_reset: resp.activity_reset,
468        }
469    }
470}
temporalio_client/async_activity_handle.rs

temporalio_client/
async_activity_handle.rs
