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

temporalio_client/
async_activity_handle.rs
