1use crate::{
4 ActivityHeartbeatResponse, ActivityIdentifier, WorkflowCancelOptions, WorkflowCountOptions,
5 WorkflowDescribeOptions, WorkflowFetchHistoryOptions, WorkflowQueryOptions,
6 WorkflowSignalOptions, WorkflowStartError, WorkflowStartOptions, WorkflowStartUpdateOptions,
7 WorkflowTerminateOptions, WorkflowUpdateWithStartOptions,
8 errors::{
9 AsyncActivityError, ClientError, WorkflowInteractionError, WorkflowQueryError,
10 WorkflowUpdateError, WorkflowUpdateWithStartError,
11 },
12 schedules::{
13 CreateScheduleOptions, ScheduleBackfill, ScheduleError, ScheduleOverlapPolicy,
14 ScheduleUpdate,
15 },
16};
17use futures_util::future::BoxFuture;
18use std::{any::Any, sync::Arc};
19use temporalio_common::{
20 data_converters::{
21 GenericPayloadConverter, PayloadConversionError, SerializationContext, TemporalSerializable,
22 },
23 protos::temporal::api::{
24 common::v1::Payload,
25 history::v1::HistoryEvent,
26 schedule::v1::ScheduleListEntry,
27 update::v1::Outcome,
28 workflow::v1::WorkflowExecutionInfo,
29 workflowservice::v1::{
30 CountWorkflowExecutionsResponse, DescribeScheduleResponse,
31 DescribeWorkflowExecutionResponse, QueryWorkflowResponse,
32 },
33 },
34};
35
36mod temporal_client_value {
37 use super::*;
38
39 pub trait Sealed {
40 fn serialize_client_payloads(
41 &self,
42 context: &SerializationContext<'_>,
43 ) -> Result<Vec<Payload>, PayloadConversionError>;
44 }
45
46 impl<T> Sealed for T
47 where
48 T: Any + TemporalSerializable + Send,
49 {
50 fn serialize_client_payloads(
51 &self,
52 context: &SerializationContext<'_>,
53 ) -> Result<Vec<Payload>, PayloadConversionError> {
54 context.converter.to_payloads(context, self)
55 }
56 }
57}
58
59pub trait TemporalClientValue: Any + Send + temporal_client_value::Sealed {
61 fn as_any(&self) -> &dyn Any;
63
64 fn as_any_mut(&mut self) -> &mut dyn Any;
66}
67
68impl<T> TemporalClientValue for T
69where
70 T: Any + TemporalSerializable + Send,
71{
72 fn as_any(&self) -> &dyn Any {
73 self
74 }
75
76 fn as_any_mut(&mut self) -> &mut dyn Any {
77 self
78 }
79}
80
81impl dyn TemporalClientValue {
82 pub(crate) fn serialize_payloads(
83 &self,
84 context: &SerializationContext<'_>,
85 ) -> Result<Vec<Payload>, PayloadConversionError> {
86 temporal_client_value::Sealed::serialize_client_payloads(self, context)
87 }
88}
89
90pub trait HasArgs {
92 fn args_ref<T: Any>(&self) -> Option<&T>;
94
95 fn args_mut<T: Any>(&mut self) -> Option<&mut T>;
97
98 fn replace_args<T>(&mut self, args: T)
100 where
101 T: TemporalSerializable + Send + 'static;
102}
103
104macro_rules! impl_with_args {
105 ($input:ty) => {
106 impl HasArgs for $input {
107 fn args_ref<T: Any>(&self) -> Option<&T> {
108 self.args.as_any().downcast_ref()
109 }
110
111 fn args_mut<T: Any>(&mut self) -> Option<&mut T> {
112 self.args.as_any_mut().downcast_mut()
113 }
114
115 fn replace_args<T>(&mut self, args: T)
116 where
117 T: TemporalSerializable + Send + 'static,
118 {
119 self.args = Box::new(args);
120 }
121 }
122 };
123}
124
125pub struct Next<'a, I, O> {
129 inner: Box<dyn FnOnce(I) -> O + Send + 'a>,
130}
131
132impl<'a, I, O> Next<'a, I, O> {
133 pub(crate) fn new(f: impl FnOnce(I) -> O + Send + 'a) -> Self {
134 Self { inner: Box::new(f) }
135 }
136
137 pub fn run(self, input: I) -> O {
139 (self.inner)(input)
140 }
141}
142
143#[non_exhaustive]
145#[derive(derive_more::Debug)]
146pub struct StartWorkflowInput {
147 pub workflow_type: String,
149 pub options: WorkflowStartOptions,
151 pub rpc_options: crate::RpcOptions,
153 #[debug(skip)]
154 args: Box<dyn TemporalClientValue>,
155}
156
157impl StartWorkflowInput {
158 pub(crate) fn new<T>(workflow_type: String, args: T, mut options: WorkflowStartOptions) -> Self
159 where
160 T: TemporalSerializable + Send + 'static,
161 {
162 let rpc_options = std::mem::take(&mut options.rpc_options);
163 Self {
164 workflow_type,
165 options,
166 rpc_options,
167 args: Box::new(args),
168 }
169 }
170
171 pub(crate) fn into_parts(
172 self,
173 ) -> (
174 String,
175 Box<dyn TemporalClientValue>,
176 WorkflowStartOptions,
177 crate::RpcOptions,
178 ) {
179 (
180 self.workflow_type,
181 self.args,
182 self.options,
183 self.rpc_options,
184 )
185 }
186}
187
188impl_with_args!(StartWorkflowInput);
189
190#[non_exhaustive]
192#[derive(derive_more::Debug)]
193pub struct SignalWithStartWorkflowInput {
194 pub workflow_type: String,
196 pub signal_name: String,
198 pub options: WorkflowStartOptions,
200 pub rpc_options: crate::RpcOptions,
202 #[debug(skip)]
205 workflow_args: Box<dyn TemporalClientValue>,
206 #[debug(skip)]
207 signal_args: Box<dyn TemporalClientValue>,
208}
209
210impl SignalWithStartWorkflowInput {
211 pub(crate) fn new<W, S>(
212 workflow_type: String,
213 workflow_args: W,
214 signal_name: String,
215 signal_args: S,
216 mut options: WorkflowStartOptions,
217 ) -> Self
218 where
219 W: TemporalSerializable + Send + 'static,
220 S: TemporalSerializable + Send + 'static,
221 {
222 let rpc_options = std::mem::take(&mut options.rpc_options);
223 Self {
224 workflow_type,
225 signal_name,
226 options,
227 rpc_options,
228 workflow_args: Box::new(workflow_args),
229 signal_args: Box::new(signal_args),
230 }
231 }
232
233 pub(crate) fn into_parts(
234 self,
235 ) -> (
236 String,
237 Box<dyn TemporalClientValue>,
238 String,
239 Box<dyn TemporalClientValue>,
240 WorkflowStartOptions,
241 crate::RpcOptions,
242 ) {
243 (
244 self.workflow_type,
245 self.workflow_args,
246 self.signal_name,
247 self.signal_args,
248 self.options,
249 self.rpc_options,
250 )
251 }
252
253 pub fn workflow_args_ref<T: Any>(&self) -> Option<&T> {
255 self.workflow_args.as_any().downcast_ref()
256 }
257
258 pub fn signal_args_ref<T: Any>(&self) -> Option<&T> {
260 self.signal_args.as_any().downcast_ref()
261 }
262
263 pub fn workflow_args_mut<T: Any>(&mut self) -> Option<&mut T> {
265 self.workflow_args.as_any_mut().downcast_mut()
266 }
267
268 pub fn signal_args_mut<T: Any>(&mut self) -> Option<&mut T> {
270 self.signal_args.as_any_mut().downcast_mut()
271 }
272
273 pub fn replace_workflow_args<T>(&mut self, args: T)
275 where
276 T: TemporalSerializable + Send + 'static,
277 {
278 self.workflow_args = Box::new(args);
279 }
280
281 pub fn replace_signal_args<T>(&mut self, args: T)
283 where
284 T: TemporalSerializable + Send + 'static,
285 {
286 self.signal_args = Box::new(args);
287 }
288}
289
290#[non_exhaustive]
292#[derive(Clone, Debug, PartialEq, Eq)]
293pub struct StartWorkflowOutput {
294 pub workflow_id: String,
296 pub run_id: String,
298}
299
300impl StartWorkflowOutput {
301 pub(crate) fn new(workflow_id: impl Into<String>, run_id: impl Into<String>) -> Self {
302 Self {
303 workflow_id: workflow_id.into(),
304 run_id: run_id.into(),
305 }
306 }
307}
308
309#[non_exhaustive]
311#[derive(Clone, Debug)]
312pub struct ListWorkflowsPageInput {
313 pub query: String,
315 pub next_page_token: Vec<u8>,
317 pub rpc_options: crate::RpcOptions,
319}
320
321#[non_exhaustive]
323#[derive(Clone, Debug)]
324pub struct ListWorkflowsPageOutput {
325 pub executions: Vec<WorkflowExecutionInfo>,
327 pub next_page_token: Vec<u8>,
329}
330
331impl ListWorkflowsPageOutput {
332 pub(crate) fn new(executions: Vec<WorkflowExecutionInfo>, next_page_token: Vec<u8>) -> Self {
333 Self {
334 executions,
335 next_page_token,
336 }
337 }
338}
339
340#[non_exhaustive]
342#[derive(Clone, Debug)]
343pub struct CountWorkflowsInput {
344 pub query: String,
346 pub options: WorkflowCountOptions,
348}
349
350#[non_exhaustive]
352#[derive(Clone, Debug)]
353pub struct CountWorkflowsOutput {
354 pub response: CountWorkflowExecutionsResponse,
356}
357
358impl CountWorkflowsOutput {
359 pub(crate) fn new(response: CountWorkflowExecutionsResponse) -> Self {
360 Self { response }
361 }
362}
363
364#[non_exhaustive]
366#[derive(Clone, Debug)]
367pub struct DescribeWorkflowInput {
368 pub workflow_id: String,
370 pub run_id: String,
372 pub options: WorkflowDescribeOptions,
374}
375
376#[non_exhaustive]
378#[derive(Clone, Debug)]
379pub struct DescribeWorkflowOutput {
380 pub response: DescribeWorkflowExecutionResponse,
382}
383
384impl DescribeWorkflowOutput {
385 pub(crate) fn new(response: DescribeWorkflowExecutionResponse) -> Self {
386 Self { response }
387 }
388}
389
390#[non_exhaustive]
392#[derive(Clone, Debug)]
393pub struct FetchWorkflowHistoryPageInput {
394 pub workflow_id: String,
396 pub run_id: String,
398 pub next_page_token: Vec<u8>,
400 pub options: WorkflowFetchHistoryOptions,
402}
403
404#[non_exhaustive]
406#[derive(Clone, Debug)]
407pub struct FetchWorkflowHistoryPageOutput {
408 pub events: Vec<HistoryEvent>,
410 pub next_page_token: Vec<u8>,
412}
413
414impl FetchWorkflowHistoryPageOutput {
415 pub(crate) fn new(events: Vec<HistoryEvent>, next_page_token: Vec<u8>) -> Self {
416 Self {
417 events,
418 next_page_token,
419 }
420 }
421}
422
423#[non_exhaustive]
425#[derive(derive_more::Debug)]
426pub struct SignalWorkflowInput {
427 pub workflow_id: String,
429 pub run_id: String,
431 pub signal_name: String,
433 pub options: WorkflowSignalOptions,
435 #[debug(skip)]
436 args: Box<dyn TemporalClientValue>,
437}
438
439impl SignalWorkflowInput {
440 pub(crate) fn new<T>(
441 workflow_id: String,
442 run_id: String,
443 signal_name: String,
444 args: T,
445 options: WorkflowSignalOptions,
446 ) -> Self
447 where
448 T: TemporalSerializable + Send + 'static,
449 {
450 Self {
451 workflow_id,
452 run_id,
453 signal_name,
454 options,
455 args: Box::new(args),
456 }
457 }
458
459 pub(crate) fn into_parts(
460 self,
461 ) -> (
462 String,
463 String,
464 String,
465 Box<dyn TemporalClientValue>,
466 WorkflowSignalOptions,
467 ) {
468 (
469 self.workflow_id,
470 self.run_id,
471 self.signal_name,
472 self.args,
473 self.options,
474 )
475 }
476}
477
478impl_with_args!(SignalWorkflowInput);
479
480#[non_exhaustive]
482#[derive(derive_more::Debug)]
483pub struct QueryWorkflowInput {
484 pub workflow_id: String,
486 pub run_id: String,
488 pub query_name: String,
490 pub options: WorkflowQueryOptions,
492 #[debug(skip)]
493 args: Box<dyn TemporalClientValue>,
494}
495
496impl QueryWorkflowInput {
497 pub(crate) fn new<T>(
498 workflow_id: String,
499 run_id: String,
500 query_name: String,
501 args: T,
502 options: WorkflowQueryOptions,
503 ) -> Self
504 where
505 T: TemporalSerializable + Send + 'static,
506 {
507 Self {
508 workflow_id,
509 run_id,
510 query_name,
511 options,
512 args: Box::new(args),
513 }
514 }
515
516 pub(crate) fn into_parts(
517 self,
518 ) -> (
519 String,
520 String,
521 String,
522 Box<dyn TemporalClientValue>,
523 WorkflowQueryOptions,
524 ) {
525 (
526 self.workflow_id,
527 self.run_id,
528 self.query_name,
529 self.args,
530 self.options,
531 )
532 }
533}
534
535impl_with_args!(QueryWorkflowInput);
536
537#[non_exhaustive]
539#[derive(Clone, Debug)]
540pub struct QueryWorkflowOutput {
541 pub response: QueryWorkflowResponse,
543}
544
545impl QueryWorkflowOutput {
546 pub(crate) fn new(response: QueryWorkflowResponse) -> Self {
547 Self { response }
548 }
549}
550
551#[non_exhaustive]
553#[derive(derive_more::Debug)]
554pub struct StartWorkflowUpdateInput {
555 pub workflow_id: String,
557 pub run_id: String,
559 pub update_name: String,
561 pub options: WorkflowStartUpdateOptions,
563 #[debug(skip)]
564 args: Box<dyn TemporalClientValue>,
565}
566
567impl StartWorkflowUpdateInput {
568 pub(crate) fn new<T>(
569 workflow_id: String,
570 run_id: String,
571 update_name: String,
572 args: T,
573 options: WorkflowStartUpdateOptions,
574 ) -> Self
575 where
576 T: TemporalSerializable + Send + 'static,
577 {
578 Self {
579 workflow_id,
580 run_id,
581 update_name,
582 options,
583 args: Box::new(args),
584 }
585 }
586
587 pub(crate) fn into_parts(
588 self,
589 ) -> (
590 String,
591 String,
592 String,
593 Box<dyn TemporalClientValue>,
594 WorkflowStartUpdateOptions,
595 ) {
596 (
597 self.workflow_id,
598 self.run_id,
599 self.update_name,
600 self.args,
601 self.options,
602 )
603 }
604}
605
606impl_with_args!(StartWorkflowUpdateInput);
607
608#[non_exhaustive]
610#[derive(Clone, Debug)]
611pub struct StartWorkflowUpdateOutput {
612 pub update_id: String,
614 pub workflow_id: String,
616 pub run_id: Option<String>,
618 pub known_outcome: Option<Outcome>,
620}
621
622impl StartWorkflowUpdateOutput {
623 pub(crate) fn new(
624 update_id: impl Into<String>,
625 workflow_id: impl Into<String>,
626 run_id: Option<String>,
627 known_outcome: Option<Outcome>,
628 ) -> Self {
629 Self {
630 update_id: update_id.into(),
631 workflow_id: workflow_id.into(),
632 run_id,
633 known_outcome,
634 }
635 }
636}
637
638#[non_exhaustive]
640#[derive(derive_more::Debug)]
641pub struct UpdateWithStartWorkflowInput {
642 pub workflow_type: String,
644 pub update_name: String,
646 pub options: WorkflowUpdateWithStartOptions,
648 pub rpc_options: crate::RpcOptions,
650 #[debug(skip)]
651 pub(crate) workflow_args: Box<dyn TemporalClientValue>,
652 #[debug(skip)]
653 pub(crate) update_args: Box<dyn TemporalClientValue>,
654}
655
656impl UpdateWithStartWorkflowInput {
657 pub(crate) fn new<WA, UA>(
658 workflow_type: String,
659 workflow_args: WA,
660 update_name: String,
661 update_args: UA,
662 mut options: WorkflowUpdateWithStartOptions,
663 ) -> Self
664 where
665 WA: TemporalSerializable + Send + 'static,
666 UA: TemporalSerializable + Send + 'static,
667 {
668 let rpc_options = std::mem::take(&mut options.rpc_options);
669 Self {
670 workflow_type,
671 update_name,
672 options,
673 rpc_options,
674 workflow_args: Box::new(workflow_args),
675 update_args: Box::new(update_args),
676 }
677 }
678
679 pub fn workflow_args_ref<T: Any>(&self) -> Option<&T> {
681 self.workflow_args.as_any().downcast_ref()
682 }
683
684 pub fn workflow_args_mut<T: Any>(&mut self) -> Option<&mut T> {
686 self.workflow_args.as_any_mut().downcast_mut()
687 }
688
689 pub fn replace_workflow_args<T>(&mut self, args: T)
691 where
692 T: TemporalSerializable + Send + 'static,
693 {
694 self.workflow_args = Box::new(args);
695 }
696
697 pub fn update_args_ref<T: Any>(&self) -> Option<&T> {
699 self.update_args.as_any().downcast_ref()
700 }
701
702 pub fn update_args_mut<T: Any>(&mut self) -> Option<&mut T> {
704 self.update_args.as_any_mut().downcast_mut()
705 }
706
707 pub fn replace_update_args<T>(&mut self, args: T)
709 where
710 T: TemporalSerializable + Send + 'static,
711 {
712 self.update_args = Box::new(args);
713 }
714}
715
716#[non_exhaustive]
718#[derive(Clone, Debug)]
719pub struct UpdateWithStartWorkflowOutput {
720 pub workflow_id: String,
722 pub update_id: String,
724 pub run_id: Option<String>,
726 pub known_outcome: Option<Outcome>,
728}
729
730impl UpdateWithStartWorkflowOutput {
731 pub(crate) fn new(
732 workflow_id: impl Into<String>,
733 update_id: impl Into<String>,
734 run_id: Option<String>,
735 known_outcome: Option<Outcome>,
736 ) -> Self {
737 Self {
738 workflow_id: workflow_id.into(),
739 update_id: update_id.into(),
740 run_id,
741 known_outcome,
742 }
743 }
744}
745
746#[non_exhaustive]
748#[derive(Clone, Debug)]
749pub struct PollWorkflowUpdateInput {
750 pub update_id: String,
752 pub workflow_id: String,
754 pub run_id: String,
756 pub rpc_options: crate::RpcOptions,
758}
759
760#[non_exhaustive]
762#[derive(Clone, Debug)]
763pub struct PollWorkflowUpdateOutput {
764 pub outcome: Outcome,
766}
767
768impl PollWorkflowUpdateOutput {
769 pub(crate) fn new(outcome: Outcome) -> Self {
770 Self { outcome }
771 }
772}
773
774#[non_exhaustive]
776#[derive(Clone, Debug)]
777pub struct CancelWorkflowInput {
778 pub workflow_id: String,
780 pub run_id: String,
782 pub first_execution_run_id: String,
784 pub options: WorkflowCancelOptions,
786}
787
788#[non_exhaustive]
790#[derive(Clone, Debug)]
791pub struct TerminateWorkflowInput {
792 pub workflow_id: String,
794 pub run_id: String,
796 pub first_execution_run_id: String,
798 pub options: WorkflowTerminateOptions,
800}
801
802#[non_exhaustive]
804#[derive(Debug)]
805pub struct CreateScheduleInput {
806 pub schedule_id: String,
808 pub options: CreateScheduleOptions,
810}
811
812#[non_exhaustive]
814#[derive(Clone, Debug, PartialEq, Eq)]
815pub struct CreateScheduleOutput {
816 pub schedule_id: String,
818}
819
820impl CreateScheduleOutput {
821 pub(crate) fn new(schedule_id: impl Into<String>) -> Self {
822 Self {
823 schedule_id: schedule_id.into(),
824 }
825 }
826}
827
828#[non_exhaustive]
830#[derive(Clone, Debug)]
831pub struct ListSchedulesPageInput {
832 pub maximum_page_size: i32,
834 pub query: String,
836 pub next_page_token: Vec<u8>,
838 pub rpc_options: crate::RpcOptions,
840}
841
842#[non_exhaustive]
844#[derive(Clone, Debug)]
845pub struct ListSchedulesPageOutput {
846 pub schedules: Vec<ScheduleListEntry>,
848 pub next_page_token: Vec<u8>,
850}
851
852impl ListSchedulesPageOutput {
853 pub(crate) fn new(schedules: Vec<ScheduleListEntry>, next_page_token: Vec<u8>) -> Self {
854 Self {
855 schedules,
856 next_page_token,
857 }
858 }
859}
860
861#[non_exhaustive]
863#[derive(Clone, Debug)]
864pub struct DescribeScheduleInput {
865 pub schedule_id: String,
867 pub rpc_options: crate::RpcOptions,
869}
870
871#[non_exhaustive]
873#[derive(Clone, Debug)]
874pub struct DescribeScheduleOutput {
875 pub response: DescribeScheduleResponse,
877}
878
879impl DescribeScheduleOutput {
880 pub(crate) fn new(response: DescribeScheduleResponse) -> Self {
881 Self { response }
882 }
883}
884
885#[non_exhaustive]
887#[derive(Clone, Debug)]
888pub struct UpdateScheduleInput {
889 pub schedule_id: String,
891 pub rpc_options: crate::RpcOptions,
893}
894
895#[non_exhaustive]
897#[derive(Clone, Debug)]
898pub struct SendScheduleUpdateInput {
899 pub schedule_id: String,
901 pub update: ScheduleUpdate,
903 pub rpc_options: crate::RpcOptions,
905}
906
907#[non_exhaustive]
909#[derive(Clone, Debug)]
910pub struct DeleteScheduleInput {
911 pub schedule_id: String,
913 pub rpc_options: crate::RpcOptions,
915}
916
917#[non_exhaustive]
919#[derive(Clone, Debug)]
920pub struct PauseScheduleInput {
921 pub schedule_id: String,
923 pub note: String,
925 pub rpc_options: crate::RpcOptions,
927}
928
929#[non_exhaustive]
931#[derive(Clone, Debug)]
932pub struct UnpauseScheduleInput {
933 pub schedule_id: String,
935 pub note: String,
937 pub rpc_options: crate::RpcOptions,
939}
940
941#[non_exhaustive]
943#[derive(Clone, Debug)]
944pub struct TriggerScheduleInput {
945 pub schedule_id: String,
947 pub overlap_policy: ScheduleOverlapPolicy,
949 pub rpc_options: crate::RpcOptions,
951}
952
953#[non_exhaustive]
955#[derive(Clone, Debug)]
956pub struct BackfillScheduleInput {
957 pub schedule_id: String,
959 pub backfills: Vec<ScheduleBackfill>,
961 pub rpc_options: crate::RpcOptions,
963}
964
965#[non_exhaustive]
967#[derive(derive_more::Debug)]
968pub struct CompleteAsyncActivityInput {
969 pub identifier: ActivityIdentifier,
971 #[debug(skip)]
972 result: Option<Box<dyn TemporalClientValue>>,
973 pub rpc_options: crate::RpcOptions,
975}
976
977impl CompleteAsyncActivityInput {
978 pub(crate) fn new<T>(
979 identifier: ActivityIdentifier,
980 result: Option<T>,
981 rpc_options: crate::RpcOptions,
982 ) -> Self
983 where
984 T: TemporalSerializable + Send + 'static,
985 {
986 Self {
987 identifier,
988 result: result.map(|value| Box::new(value) as Box<dyn TemporalClientValue>),
989 rpc_options,
990 }
991 }
992
993 pub(crate) fn into_parts(
994 self,
995 ) -> (
996 ActivityIdentifier,
997 Option<Box<dyn TemporalClientValue>>,
998 crate::RpcOptions,
999 ) {
1000 (self.identifier, self.result, self.rpc_options)
1001 }
1002
1003 pub fn result_ref<T: Any>(&self) -> Option<&T> {
1005 self.result
1006 .as_ref()
1007 .and_then(|result| result.as_any().downcast_ref())
1008 }
1009
1010 pub fn result_mut<T: Any>(&mut self) -> Option<&mut T> {
1012 self.result
1013 .as_mut()
1014 .and_then(|result| result.as_any_mut().downcast_mut())
1015 }
1016
1017 pub fn replace_result<T>(&mut self, result: Option<T>)
1019 where
1020 T: TemporalSerializable + Send + 'static,
1021 {
1022 self.result = result.map(|value| Box::new(value) as Box<dyn TemporalClientValue>);
1023 }
1024}
1025
1026#[non_exhaustive]
1028#[derive(derive_more::Debug)]
1029pub struct FailAsyncActivityInput {
1030 pub identifier: ActivityIdentifier,
1032 pub failure: temporalio_common::error::ApplicationFailure,
1034 #[debug(skip)]
1035 last_heartbeat_details: Option<Box<dyn TemporalClientValue>>,
1036 pub rpc_options: crate::RpcOptions,
1038}
1039
1040impl FailAsyncActivityInput {
1041 pub(crate) fn new<T>(
1042 identifier: ActivityIdentifier,
1043 failure: temporalio_common::error::ApplicationFailure,
1044 last_heartbeat_details: Option<T>,
1045 rpc_options: crate::RpcOptions,
1046 ) -> Self
1047 where
1048 T: TemporalSerializable + Send + 'static,
1049 {
1050 Self {
1051 identifier,
1052 failure,
1053 last_heartbeat_details: last_heartbeat_details
1054 .map(|value| Box::new(value) as Box<dyn TemporalClientValue>),
1055 rpc_options,
1056 }
1057 }
1058
1059 pub(crate) fn into_parts(
1060 self,
1061 ) -> (
1062 ActivityIdentifier,
1063 temporalio_common::error::ApplicationFailure,
1064 Option<Box<dyn TemporalClientValue>>,
1065 crate::RpcOptions,
1066 ) {
1067 (
1068 self.identifier,
1069 self.failure,
1070 self.last_heartbeat_details,
1071 self.rpc_options,
1072 )
1073 }
1074
1075 pub fn last_heartbeat_details_ref<T: Any>(&self) -> Option<&T> {
1077 self.last_heartbeat_details
1078 .as_ref()
1079 .and_then(|details| details.as_any().downcast_ref())
1080 }
1081
1082 pub fn last_heartbeat_details_mut<T: Any>(&mut self) -> Option<&mut T> {
1084 self.last_heartbeat_details
1085 .as_mut()
1086 .and_then(|details| details.as_any_mut().downcast_mut())
1087 }
1088
1089 pub fn replace_last_heartbeat_details<T>(&mut self, details: Option<T>)
1091 where
1092 T: TemporalSerializable + Send + 'static,
1093 {
1094 self.last_heartbeat_details =
1095 details.map(|value| Box::new(value) as Box<dyn TemporalClientValue>);
1096 }
1097}
1098
1099#[non_exhaustive]
1101#[derive(derive_more::Debug)]
1102pub struct ReportAsyncActivityCancellationInput {
1103 pub identifier: ActivityIdentifier,
1105 #[debug(skip)]
1106 details: Option<Box<dyn TemporalClientValue>>,
1107 pub rpc_options: crate::RpcOptions,
1109}
1110
1111impl ReportAsyncActivityCancellationInput {
1112 pub(crate) fn new<T>(
1113 identifier: ActivityIdentifier,
1114 details: Option<T>,
1115 rpc_options: crate::RpcOptions,
1116 ) -> Self
1117 where
1118 T: TemporalSerializable + Send + 'static,
1119 {
1120 Self {
1121 identifier,
1122 details: details.map(|value| Box::new(value) as Box<dyn TemporalClientValue>),
1123 rpc_options,
1124 }
1125 }
1126
1127 pub(crate) fn into_parts(
1128 self,
1129 ) -> (
1130 ActivityIdentifier,
1131 Option<Box<dyn TemporalClientValue>>,
1132 crate::RpcOptions,
1133 ) {
1134 (self.identifier, self.details, self.rpc_options)
1135 }
1136
1137 pub fn details_ref<T: Any>(&self) -> Option<&T> {
1139 self.details
1140 .as_ref()
1141 .and_then(|details| details.as_any().downcast_ref())
1142 }
1143
1144 pub fn details_mut<T: Any>(&mut self) -> Option<&mut T> {
1146 self.details
1147 .as_mut()
1148 .and_then(|details| details.as_any_mut().downcast_mut())
1149 }
1150
1151 pub fn replace_details<T>(&mut self, details: Option<T>)
1153 where
1154 T: TemporalSerializable + Send + 'static,
1155 {
1156 self.details = details.map(|value| Box::new(value) as Box<dyn TemporalClientValue>);
1157 }
1158}
1159
1160#[non_exhaustive]
1162#[derive(derive_more::Debug)]
1163pub struct HeartbeatAsyncActivityInput {
1164 pub identifier: ActivityIdentifier,
1166 #[debug(skip)]
1167 details: Option<Box<dyn TemporalClientValue>>,
1168 pub rpc_options: crate::RpcOptions,
1170}
1171
1172impl HeartbeatAsyncActivityInput {
1173 pub(crate) fn new<T>(
1174 identifier: ActivityIdentifier,
1175 details: Option<T>,
1176 rpc_options: crate::RpcOptions,
1177 ) -> Self
1178 where
1179 T: TemporalSerializable + Send + 'static,
1180 {
1181 Self {
1182 identifier,
1183 details: details.map(|value| Box::new(value) as Box<dyn TemporalClientValue>),
1184 rpc_options,
1185 }
1186 }
1187
1188 pub(crate) fn into_parts(
1189 self,
1190 ) -> (
1191 ActivityIdentifier,
1192 Option<Box<dyn TemporalClientValue>>,
1193 crate::RpcOptions,
1194 ) {
1195 (self.identifier, self.details, self.rpc_options)
1196 }
1197
1198 pub fn details_ref<T: Any>(&self) -> Option<&T> {
1200 self.details
1201 .as_ref()
1202 .and_then(|details| details.as_any().downcast_ref())
1203 }
1204
1205 pub fn details_mut<T: Any>(&mut self) -> Option<&mut T> {
1207 self.details
1208 .as_mut()
1209 .and_then(|details| details.as_any_mut().downcast_mut())
1210 }
1211
1212 pub fn replace_details<T>(&mut self, details: Option<T>)
1214 where
1215 T: TemporalSerializable + Send + 'static,
1216 {
1217 self.details = details.map(|value| Box::new(value) as Box<dyn TemporalClientValue>);
1218 }
1219}
1220
1221pub trait ClientInterceptor: Send + Sync + 'static {
1260 fn start_workflow<'a>(
1262 &'a self,
1263 input: StartWorkflowInput,
1264 next: Next<
1265 'a,
1266 StartWorkflowInput,
1267 BoxFuture<'a, Result<StartWorkflowOutput, WorkflowStartError>>,
1268 >,
1269 ) -> BoxFuture<'a, Result<StartWorkflowOutput, WorkflowStartError>> {
1270 next.run(input)
1271 }
1272
1273 fn signal_with_start_workflow<'a>(
1275 &'a self,
1276 input: SignalWithStartWorkflowInput,
1277 next: Next<
1278 'a,
1279 SignalWithStartWorkflowInput,
1280 BoxFuture<'a, Result<StartWorkflowOutput, WorkflowStartError>>,
1281 >,
1282 ) -> BoxFuture<'a, Result<StartWorkflowOutput, WorkflowStartError>> {
1283 next.run(input)
1284 }
1285
1286 fn list_workflows_page<'a>(
1288 &'a self,
1289 input: ListWorkflowsPageInput,
1290 next: Next<
1291 'a,
1292 ListWorkflowsPageInput,
1293 BoxFuture<'a, Result<ListWorkflowsPageOutput, ClientError>>,
1294 >,
1295 ) -> BoxFuture<'a, Result<ListWorkflowsPageOutput, ClientError>> {
1296 next.run(input)
1297 }
1298
1299 fn count_workflows<'a>(
1301 &'a self,
1302 input: CountWorkflowsInput,
1303 next: Next<
1304 'a,
1305 CountWorkflowsInput,
1306 BoxFuture<'a, Result<CountWorkflowsOutput, ClientError>>,
1307 >,
1308 ) -> BoxFuture<'a, Result<CountWorkflowsOutput, ClientError>> {
1309 next.run(input)
1310 }
1311
1312 fn describe_workflow<'a>(
1314 &'a self,
1315 input: DescribeWorkflowInput,
1316 next: Next<
1317 'a,
1318 DescribeWorkflowInput,
1319 BoxFuture<'a, Result<DescribeWorkflowOutput, WorkflowInteractionError>>,
1320 >,
1321 ) -> BoxFuture<'a, Result<DescribeWorkflowOutput, WorkflowInteractionError>> {
1322 next.run(input)
1323 }
1324
1325 fn fetch_workflow_history_page<'a>(
1327 &'a self,
1328 input: FetchWorkflowHistoryPageInput,
1329 next: Next<
1330 'a,
1331 FetchWorkflowHistoryPageInput,
1332 BoxFuture<'a, Result<FetchWorkflowHistoryPageOutput, WorkflowInteractionError>>,
1333 >,
1334 ) -> BoxFuture<'a, Result<FetchWorkflowHistoryPageOutput, WorkflowInteractionError>> {
1335 next.run(input)
1336 }
1337
1338 fn signal_workflow<'a>(
1340 &'a self,
1341 input: SignalWorkflowInput,
1342 next: Next<'a, SignalWorkflowInput, BoxFuture<'a, Result<(), WorkflowInteractionError>>>,
1343 ) -> BoxFuture<'a, Result<(), WorkflowInteractionError>> {
1344 next.run(input)
1345 }
1346
1347 fn query_workflow<'a>(
1349 &'a self,
1350 input: QueryWorkflowInput,
1351 next: Next<
1352 'a,
1353 QueryWorkflowInput,
1354 BoxFuture<'a, Result<QueryWorkflowOutput, WorkflowQueryError>>,
1355 >,
1356 ) -> BoxFuture<'a, Result<QueryWorkflowOutput, WorkflowQueryError>> {
1357 next.run(input)
1358 }
1359
1360 fn start_workflow_update<'a>(
1362 &'a self,
1363 input: StartWorkflowUpdateInput,
1364 next: Next<
1365 'a,
1366 StartWorkflowUpdateInput,
1367 BoxFuture<'a, Result<StartWorkflowUpdateOutput, WorkflowUpdateError>>,
1368 >,
1369 ) -> BoxFuture<'a, Result<StartWorkflowUpdateOutput, WorkflowUpdateError>> {
1370 next.run(input)
1371 }
1372
1373 fn update_with_start_workflow<'a>(
1375 &'a self,
1376 input: UpdateWithStartWorkflowInput,
1377 next: Next<
1378 'a,
1379 UpdateWithStartWorkflowInput,
1380 BoxFuture<'a, Result<UpdateWithStartWorkflowOutput, WorkflowUpdateWithStartError>>,
1381 >,
1382 ) -> BoxFuture<'a, Result<UpdateWithStartWorkflowOutput, WorkflowUpdateWithStartError>> {
1383 next.run(input)
1384 }
1385
1386 fn poll_workflow_update<'a>(
1388 &'a self,
1389 input: PollWorkflowUpdateInput,
1390 next: Next<
1391 'a,
1392 PollWorkflowUpdateInput,
1393 BoxFuture<'a, Result<PollWorkflowUpdateOutput, WorkflowUpdateError>>,
1394 >,
1395 ) -> BoxFuture<'a, Result<PollWorkflowUpdateOutput, WorkflowUpdateError>> {
1396 next.run(input)
1397 }
1398
1399 fn cancel_workflow<'a>(
1401 &'a self,
1402 input: CancelWorkflowInput,
1403 next: Next<'a, CancelWorkflowInput, BoxFuture<'a, Result<(), WorkflowInteractionError>>>,
1404 ) -> BoxFuture<'a, Result<(), WorkflowInteractionError>> {
1405 next.run(input)
1406 }
1407
1408 fn terminate_workflow<'a>(
1410 &'a self,
1411 input: TerminateWorkflowInput,
1412 next: Next<'a, TerminateWorkflowInput, BoxFuture<'a, Result<(), WorkflowInteractionError>>>,
1413 ) -> BoxFuture<'a, Result<(), WorkflowInteractionError>> {
1414 next.run(input)
1415 }
1416
1417 fn create_schedule<'a>(
1419 &'a self,
1420 input: CreateScheduleInput,
1421 next: Next<
1422 'a,
1423 CreateScheduleInput,
1424 BoxFuture<'a, Result<CreateScheduleOutput, ScheduleError>>,
1425 >,
1426 ) -> BoxFuture<'a, Result<CreateScheduleOutput, ScheduleError>> {
1427 next.run(input)
1428 }
1429
1430 fn list_schedules_page<'a>(
1432 &'a self,
1433 input: ListSchedulesPageInput,
1434 next: Next<
1435 'a,
1436 ListSchedulesPageInput,
1437 BoxFuture<'a, Result<ListSchedulesPageOutput, ScheduleError>>,
1438 >,
1439 ) -> BoxFuture<'a, Result<ListSchedulesPageOutput, ScheduleError>> {
1440 next.run(input)
1441 }
1442
1443 fn describe_schedule<'a>(
1445 &'a self,
1446 input: DescribeScheduleInput,
1447 next: Next<
1448 'a,
1449 DescribeScheduleInput,
1450 BoxFuture<'a, Result<DescribeScheduleOutput, ScheduleError>>,
1451 >,
1452 ) -> BoxFuture<'a, Result<DescribeScheduleOutput, ScheduleError>> {
1453 next.run(input)
1454 }
1455
1456 fn update_schedule<'a>(
1458 &'a self,
1459 input: UpdateScheduleInput,
1460 next: Next<'a, UpdateScheduleInput, BoxFuture<'a, Result<(), ScheduleError>>>,
1461 ) -> BoxFuture<'a, Result<(), ScheduleError>> {
1462 next.run(input)
1463 }
1464
1465 fn send_schedule_update<'a>(
1467 &'a self,
1468 input: SendScheduleUpdateInput,
1469 next: Next<'a, SendScheduleUpdateInput, BoxFuture<'a, Result<(), ScheduleError>>>,
1470 ) -> BoxFuture<'a, Result<(), ScheduleError>> {
1471 next.run(input)
1472 }
1473
1474 fn delete_schedule<'a>(
1476 &'a self,
1477 input: DeleteScheduleInput,
1478 next: Next<'a, DeleteScheduleInput, BoxFuture<'a, Result<(), ScheduleError>>>,
1479 ) -> BoxFuture<'a, Result<(), ScheduleError>> {
1480 next.run(input)
1481 }
1482
1483 fn pause_schedule<'a>(
1485 &'a self,
1486 input: PauseScheduleInput,
1487 next: Next<'a, PauseScheduleInput, BoxFuture<'a, Result<(), ScheduleError>>>,
1488 ) -> BoxFuture<'a, Result<(), ScheduleError>> {
1489 next.run(input)
1490 }
1491
1492 fn unpause_schedule<'a>(
1494 &'a self,
1495 input: UnpauseScheduleInput,
1496 next: Next<'a, UnpauseScheduleInput, BoxFuture<'a, Result<(), ScheduleError>>>,
1497 ) -> BoxFuture<'a, Result<(), ScheduleError>> {
1498 next.run(input)
1499 }
1500
1501 fn trigger_schedule<'a>(
1503 &'a self,
1504 input: TriggerScheduleInput,
1505 next: Next<'a, TriggerScheduleInput, BoxFuture<'a, Result<(), ScheduleError>>>,
1506 ) -> BoxFuture<'a, Result<(), ScheduleError>> {
1507 next.run(input)
1508 }
1509
1510 fn backfill_schedule<'a>(
1512 &'a self,
1513 input: BackfillScheduleInput,
1514 next: Next<'a, BackfillScheduleInput, BoxFuture<'a, Result<(), ScheduleError>>>,
1515 ) -> BoxFuture<'a, Result<(), ScheduleError>> {
1516 next.run(input)
1517 }
1518
1519 fn complete_async_activity<'a>(
1521 &'a self,
1522 input: CompleteAsyncActivityInput,
1523 next: Next<'a, CompleteAsyncActivityInput, BoxFuture<'a, Result<(), AsyncActivityError>>>,
1524 ) -> BoxFuture<'a, Result<(), AsyncActivityError>> {
1525 next.run(input)
1526 }
1527
1528 fn fail_async_activity<'a>(
1530 &'a self,
1531 input: FailAsyncActivityInput,
1532 next: Next<'a, FailAsyncActivityInput, BoxFuture<'a, Result<(), AsyncActivityError>>>,
1533 ) -> BoxFuture<'a, Result<(), AsyncActivityError>> {
1534 next.run(input)
1535 }
1536
1537 fn report_async_activity_cancellation<'a>(
1539 &'a self,
1540 input: ReportAsyncActivityCancellationInput,
1541 next: Next<
1542 'a,
1543 ReportAsyncActivityCancellationInput,
1544 BoxFuture<'a, Result<(), AsyncActivityError>>,
1545 >,
1546 ) -> BoxFuture<'a, Result<(), AsyncActivityError>> {
1547 next.run(input)
1548 }
1549
1550 fn heartbeat_async_activity<'a>(
1552 &'a self,
1553 input: HeartbeatAsyncActivityInput,
1554 next: Next<
1555 'a,
1556 HeartbeatAsyncActivityInput,
1557 BoxFuture<'a, Result<ActivityHeartbeatResponse, AsyncActivityError>>,
1558 >,
1559 ) -> BoxFuture<'a, Result<ActivityHeartbeatResponse, AsyncActivityError>> {
1560 next.run(input)
1561 }
1562}
1563
1564macro_rules! interceptor_chain {
1565 ($fn_name:ident, $method:ident, $input:ty, $output:ty) => {
1566 pub(crate) fn $fn_name<'a>(
1567 interceptors: &'a [Arc<dyn ClientInterceptor>],
1568 input: $input,
1569 terminal: Next<'a, $input, $output>,
1570 ) -> $output {
1571 if let Some((interceptor, remaining)) = interceptors.split_first() {
1572 let next = Next::new(move |input| $fn_name(remaining, input, terminal));
1573 interceptor.$method(input, next)
1574 } else {
1575 terminal.run(input)
1576 }
1577 }
1578 };
1579}
1580
1581interceptor_chain!(
1582 call_start_workflow,
1583 start_workflow,
1584 StartWorkflowInput,
1585 BoxFuture<'a, Result<StartWorkflowOutput, WorkflowStartError>>
1586);
1587
1588interceptor_chain!(
1589 call_signal_with_start_workflow,
1590 signal_with_start_workflow,
1591 SignalWithStartWorkflowInput,
1592 BoxFuture<'a, Result<StartWorkflowOutput, WorkflowStartError>>
1593);
1594
1595interceptor_chain!(
1596 call_list_workflows_page,
1597 list_workflows_page,
1598 ListWorkflowsPageInput,
1599 BoxFuture<'a, Result<ListWorkflowsPageOutput, ClientError>>
1600);
1601
1602interceptor_chain!(
1603 call_count_workflows,
1604 count_workflows,
1605 CountWorkflowsInput,
1606 BoxFuture<'a, Result<CountWorkflowsOutput, ClientError>>
1607);
1608
1609interceptor_chain!(
1610 call_describe_workflow,
1611 describe_workflow,
1612 DescribeWorkflowInput,
1613 BoxFuture<'a, Result<DescribeWorkflowOutput, WorkflowInteractionError>>
1614);
1615
1616interceptor_chain!(
1617 call_fetch_workflow_history_page,
1618 fetch_workflow_history_page,
1619 FetchWorkflowHistoryPageInput,
1620 BoxFuture<'a, Result<FetchWorkflowHistoryPageOutput, WorkflowInteractionError>>
1621);
1622
1623interceptor_chain!(
1624 call_signal_workflow,
1625 signal_workflow,
1626 SignalWorkflowInput,
1627 BoxFuture<'a, Result<(), WorkflowInteractionError>>
1628);
1629
1630interceptor_chain!(
1631 call_query_workflow,
1632 query_workflow,
1633 QueryWorkflowInput,
1634 BoxFuture<'a, Result<QueryWorkflowOutput, WorkflowQueryError>>
1635);
1636
1637interceptor_chain!(
1638 call_start_workflow_update,
1639 start_workflow_update,
1640 StartWorkflowUpdateInput,
1641 BoxFuture<'a, Result<StartWorkflowUpdateOutput, WorkflowUpdateError>>
1642);
1643
1644interceptor_chain!(
1645 call_update_with_start_workflow,
1646 update_with_start_workflow,
1647 UpdateWithStartWorkflowInput,
1648 BoxFuture<'a, Result<UpdateWithStartWorkflowOutput, WorkflowUpdateWithStartError>>
1649);
1650
1651interceptor_chain!(
1652 call_poll_workflow_update,
1653 poll_workflow_update,
1654 PollWorkflowUpdateInput,
1655 BoxFuture<'a, Result<PollWorkflowUpdateOutput, WorkflowUpdateError>>
1656);
1657
1658interceptor_chain!(
1659 call_cancel_workflow,
1660 cancel_workflow,
1661 CancelWorkflowInput,
1662 BoxFuture<'a, Result<(), WorkflowInteractionError>>
1663);
1664
1665interceptor_chain!(
1666 call_terminate_workflow,
1667 terminate_workflow,
1668 TerminateWorkflowInput,
1669 BoxFuture<'a, Result<(), WorkflowInteractionError>>
1670);
1671
1672interceptor_chain!(
1673 call_create_schedule,
1674 create_schedule,
1675 CreateScheduleInput,
1676 BoxFuture<'a, Result<CreateScheduleOutput, ScheduleError>>
1677);
1678
1679interceptor_chain!(
1680 call_list_schedules_page,
1681 list_schedules_page,
1682 ListSchedulesPageInput,
1683 BoxFuture<'a, Result<ListSchedulesPageOutput, ScheduleError>>
1684);
1685
1686interceptor_chain!(
1687 call_describe_schedule,
1688 describe_schedule,
1689 DescribeScheduleInput,
1690 BoxFuture<'a, Result<DescribeScheduleOutput, ScheduleError>>
1691);
1692
1693interceptor_chain!(
1694 call_update_schedule,
1695 update_schedule,
1696 UpdateScheduleInput,
1697 BoxFuture<'a, Result<(), ScheduleError>>
1698);
1699
1700interceptor_chain!(
1701 call_send_schedule_update,
1702 send_schedule_update,
1703 SendScheduleUpdateInput,
1704 BoxFuture<'a, Result<(), ScheduleError>>
1705);
1706
1707interceptor_chain!(
1708 call_delete_schedule,
1709 delete_schedule,
1710 DeleteScheduleInput,
1711 BoxFuture<'a, Result<(), ScheduleError>>
1712);
1713
1714interceptor_chain!(
1715 call_pause_schedule,
1716 pause_schedule,
1717 PauseScheduleInput,
1718 BoxFuture<'a, Result<(), ScheduleError>>
1719);
1720
1721interceptor_chain!(
1722 call_unpause_schedule,
1723 unpause_schedule,
1724 UnpauseScheduleInput,
1725 BoxFuture<'a, Result<(), ScheduleError>>
1726);
1727
1728interceptor_chain!(
1729 call_trigger_schedule,
1730 trigger_schedule,
1731 TriggerScheduleInput,
1732 BoxFuture<'a, Result<(), ScheduleError>>
1733);
1734
1735interceptor_chain!(
1736 call_backfill_schedule,
1737 backfill_schedule,
1738 BackfillScheduleInput,
1739 BoxFuture<'a, Result<(), ScheduleError>>
1740);
1741
1742interceptor_chain!(
1743 call_complete_async_activity,
1744 complete_async_activity,
1745 CompleteAsyncActivityInput,
1746 BoxFuture<'a, Result<(), AsyncActivityError>>
1747);
1748
1749interceptor_chain!(
1750 call_fail_async_activity,
1751 fail_async_activity,
1752 FailAsyncActivityInput,
1753 BoxFuture<'a, Result<(), AsyncActivityError>>
1754);
1755
1756interceptor_chain!(
1757 call_report_async_activity_cancellation,
1758 report_async_activity_cancellation,
1759 ReportAsyncActivityCancellationInput,
1760 BoxFuture<'a, Result<(), AsyncActivityError>>
1761);
1762
1763interceptor_chain!(
1764 call_heartbeat_async_activity,
1765 heartbeat_async_activity,
1766 HeartbeatAsyncActivityInput,
1767 BoxFuture<'a, Result<ActivityHeartbeatResponse, AsyncActivityError>>
1768);