1pub mod constants;
6mod task_token;
7pub mod utilities;
9pub use task_token::TaskToken;
10
11pub static ENCODING_PAYLOAD_KEY: &str = "encoding";
13pub static JSON_ENCODING_VAL: &str = "json/plain";
15pub static PATCHED_MARKER_DETAILS_KEY: &str = "patch-data";
17pub static VERSION_SEARCH_ATTR_KEY: &str = "TemporalChangeVersion";
19
20macro_rules! include_proto_with_serde {
21 ($pkg:tt) => {
22 tonic::include_proto!($pkg);
23
24 #[cfg(feature = "serde_serialize")]
25 include!(concat!(env!("OUT_DIR"), concat!("/", $pkg, ".serde.rs")));
26 };
27}
28
29#[allow(
30 clippy::large_enum_variant,
31 clippy::derive_partial_eq_without_eq,
32 clippy::reserve_after_initialization
33)]
34#[allow(missing_docs)]
36pub mod coresdk {
37 tonic::include_proto!("coresdk");
39 pub use self::sdk_helpers::*;
40 mod sdk_helpers {
41 use super::*;
42
43 use crate::protos::{
44 ENCODING_PAYLOAD_KEY, JSON_ENCODING_VAL,
45 temporal::api::{
46 common::v1::{Payload, Payloads, RetryPolicy, WorkflowExecution},
47 enums::v1::{
48 ApplicationErrorCategory, TimeoutType, VersioningBehavior,
49 WorkflowTaskFailedCause,
50 },
51 failure::v1::{
52 ActivityFailureInfo, ApplicationFailureInfo, Failure, TimeoutFailureInfo,
53 failure::FailureInfo,
54 },
55 workflowservice::v1::PollActivityTaskQueueResponse,
56 },
57 };
58 use activity_task::ActivityTask;
59 use serde::{Deserialize, Serialize};
60 use std::{
61 collections::HashMap,
62 convert::TryFrom,
63 fmt::{Display, Formatter},
64 iter::FromIterator,
65 };
66 use workflow_activation::{WorkflowActivationJob, workflow_activation_job};
67 use workflow_commands::{WorkflowCommand, workflow_command, workflow_command::Variant};
68 use workflow_completion::{WorkflowActivationCompletion, workflow_activation_completion};
69
70 pub type HistoryEventId = i64;
71
72 impl From<workflow_activation_job::Variant> for WorkflowActivationJob {
73 fn from(a: workflow_activation_job::Variant) -> Self {
74 Self { variant: Some(a) }
75 }
76 }
77
78 impl From<Vec<WorkflowCommand>> for workflow_completion::Success {
79 fn from(v: Vec<WorkflowCommand>) -> Self {
80 Self {
81 commands: v,
82 used_internal_flags: vec![],
83 versioning_behavior: VersioningBehavior::Unspecified.into(),
84 }
85 }
86 }
87
88 impl From<workflow_command::Variant> for WorkflowCommand {
89 fn from(v: workflow_command::Variant) -> Self {
90 Self {
91 variant: Some(v),
92 user_metadata: None,
93 }
94 }
95 }
96
97 impl workflow_completion::Success {
98 pub fn from_variants(cmds: Vec<Variant>) -> Self {
99 let cmds: Vec<_> = cmds.into_iter().map(|c| c.into()).collect();
100 cmds.into()
101 }
102 }
103
104 impl WorkflowActivationCompletion {
105 pub fn empty(run_id: impl Into<String>) -> Self {
107 let success = workflow_completion::Success::from_variants(vec![]);
108 Self {
109 run_id: run_id.into(),
110 status: Some(workflow_activation_completion::Status::Successful(success)),
111 }
112 }
113
114 pub fn from_cmds(
116 run_id: impl Into<String>,
117 cmds: Vec<workflow_command::Variant>,
118 ) -> Self {
119 let success = workflow_completion::Success::from_variants(cmds);
120 Self {
121 run_id: run_id.into(),
122 status: Some(workflow_activation_completion::Status::Successful(success)),
123 }
124 }
125
126 pub fn from_cmd(run_id: impl Into<String>, cmd: workflow_command::Variant) -> Self {
128 let success = workflow_completion::Success::from_variants(vec![cmd]);
129 Self {
130 run_id: run_id.into(),
131 status: Some(workflow_activation_completion::Status::Successful(success)),
132 }
133 }
134
135 pub fn fail(
136 run_id: impl Into<String>,
137 failure: Failure,
138 cause: Option<WorkflowTaskFailedCause>,
139 ) -> Self {
140 Self {
141 run_id: run_id.into(),
142 status: Some(workflow_activation_completion::Status::Failed(
143 workflow_completion::Failure {
144 failure: Some(failure),
145 force_cause: cause.unwrap_or(WorkflowTaskFailedCause::Unspecified)
146 as i32,
147 },
148 )),
149 }
150 }
151
152 pub fn has_execution_ending(&self) -> bool {
155 self.has_complete_workflow_execution()
156 || self.has_fail_execution()
157 || self.has_continue_as_new()
158 || self.has_cancel_workflow_execution()
159 }
160
161 pub fn has_fail_execution(&self) -> bool {
163 if let Some(workflow_activation_completion::Status::Successful(s)) = &self.status {
164 return s.commands.iter().any(|wfc| {
165 matches!(
166 wfc,
167 WorkflowCommand {
168 variant: Some(workflow_command::Variant::FailWorkflowExecution(_)),
169 ..
170 }
171 )
172 });
173 }
174 false
175 }
176
177 pub fn has_cancel_workflow_execution(&self) -> bool {
179 if let Some(workflow_activation_completion::Status::Successful(s)) = &self.status {
180 return s.commands.iter().any(|wfc| {
181 matches!(
182 wfc,
183 WorkflowCommand {
184 variant: Some(workflow_command::Variant::CancelWorkflowExecution(
185 _
186 )),
187 ..
188 }
189 )
190 });
191 }
192 false
193 }
194
195 pub fn has_continue_as_new(&self) -> bool {
197 if let Some(workflow_activation_completion::Status::Successful(s)) = &self.status {
198 return s.commands.iter().any(|wfc| {
199 matches!(
200 wfc,
201 WorkflowCommand {
202 variant: Some(
203 workflow_command::Variant::ContinueAsNewWorkflowExecution(_)
204 ),
205 ..
206 }
207 )
208 });
209 }
210 false
211 }
212
213 pub fn has_complete_workflow_execution(&self) -> bool {
215 self.complete_workflow_execution_value().is_some()
216 }
217
218 pub fn complete_workflow_execution_value(&self) -> Option<&Payload> {
220 if let Some(workflow_activation_completion::Status::Successful(s)) = &self.status {
221 s.commands.iter().find_map(|wfc| match wfc {
222 WorkflowCommand {
223 variant: Some(workflow_command::Variant::CompleteWorkflowExecution(v)),
224 ..
225 } => v.result.as_ref(),
226 _ => None,
227 })
228 } else {
229 None
230 }
231 }
232
233 pub fn is_empty(&self) -> bool {
235 if let Some(workflow_activation_completion::Status::Successful(s)) = &self.status {
236 return s.commands.is_empty();
237 }
238 false
239 }
240
241 pub fn add_internal_flags(&mut self, patch: u32) {
242 if let Some(workflow_activation_completion::Status::Successful(s)) =
243 &mut self.status
244 {
245 s.used_internal_flags.push(patch);
246 }
247 }
248 }
249
250 pub trait IntoCompletion {
252 fn into_completion(self, run_id: String) -> WorkflowActivationCompletion;
254 }
255
256 impl IntoCompletion for workflow_command::Variant {
257 fn into_completion(self, run_id: String) -> WorkflowActivationCompletion {
258 WorkflowActivationCompletion::from_cmd(run_id, self)
259 }
260 }
261
262 impl<I, V> IntoCompletion for I
263 where
264 I: IntoIterator<Item = V>,
265 V: Into<WorkflowCommand>,
266 {
267 fn into_completion(self, run_id: String) -> WorkflowActivationCompletion {
268 let success = self.into_iter().map(Into::into).collect::<Vec<_>>().into();
269 WorkflowActivationCompletion {
270 run_id,
271 status: Some(workflow_activation_completion::Status::Successful(success)),
272 }
273 }
274 }
275
276 impl Display for WorkflowActivationCompletion {
277 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
278 write!(
279 f,
280 "WorkflowActivationCompletion(run_id: {}, status: ",
281 &self.run_id
282 )?;
283 match &self.status {
284 None => write!(f, "empty")?,
285 Some(s) => write!(f, "{s}")?,
286 };
287 write!(f, ")")
288 }
289 }
290
291 impl Display for workflow_activation_completion::Status {
292 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
293 match self {
294 workflow_activation_completion::Status::Successful(
295 workflow_completion::Success { commands, .. },
296 ) => {
297 write!(f, "Success(")?;
298 let mut written = 0;
299 for c in commands {
300 write!(f, "{c} ")?;
301 written += 1;
302 if written >= 10 && written < commands.len() {
303 write!(f, "... {} more", commands.len() - written)?;
304 break;
305 }
306 }
307 write!(f, ")")
308 }
309 workflow_activation_completion::Status::Failed(_) => {
310 write!(f, "Failed")
311 }
312 }
313 }
314 }
315
316 impl ActivityTask {
317 pub fn start_from_poll_resp(r: PollActivityTaskQueueResponse) -> Self {
318 let (workflow_id, run_id) = r
319 .workflow_execution
320 .map(|we| (we.workflow_id, we.run_id))
321 .unwrap_or_default();
322 Self {
323 task_token: r.task_token,
324 variant: Some(activity_task::activity_task::Variant::Start(
325 activity_task::Start {
326 workflow_namespace: r.workflow_namespace,
327 workflow_type: r
328 .workflow_type
329 .map_or_else(|| "".to_string(), |wt| wt.name),
330 workflow_execution: Some(WorkflowExecution {
331 workflow_id,
332 run_id,
333 }),
334 activity_id: r.activity_id,
335 activity_type: r
336 .activity_type
337 .map_or_else(|| "".to_string(), |at| at.name),
338 header_fields: r.header.map(Into::into).unwrap_or_default(),
339 input: Vec::from_payloads(r.input),
340 heartbeat_details: Vec::from_payloads(r.heartbeat_details),
341 scheduled_time: r.scheduled_time,
342 current_attempt_scheduled_time: r.current_attempt_scheduled_time,
343 started_time: r.started_time,
344 attempt: r.attempt as u32,
345 schedule_to_close_timeout: r.schedule_to_close_timeout,
346 start_to_close_timeout: r.start_to_close_timeout,
347 heartbeat_timeout: r.heartbeat_timeout,
348 retry_policy: r.retry_policy.map(fix_retry_policy),
349 priority: r.priority,
350 is_local: false,
351 run_id: r.activity_run_id,
352 },
353 )),
354 }
355 }
356 }
357
358 impl Failure {
359 pub fn is_timeout(
360 &self,
361 ) -> Option<crate::protos::temporal::api::enums::v1::TimeoutType> {
362 match &self.failure_info {
363 Some(FailureInfo::TimeoutFailureInfo(ti)) => Some(ti.timeout_type()),
364 _ => {
365 if let Some(c) = &self.cause {
366 c.is_timeout()
367 } else {
368 None
369 }
370 }
371 }
372 }
373
374 pub fn application_failure(message: String, non_retryable: bool) -> Self {
375 Self {
376 message,
377 failure_info: Some(FailureInfo::ApplicationFailureInfo(
378 ApplicationFailureInfo {
379 non_retryable,
380 ..Default::default()
381 },
382 )),
383 ..Default::default()
384 }
385 }
386
387 pub fn application_failure_from_error(ae: anyhow::Error, non_retryable: bool) -> Self {
388 Self {
389 failure_info: Some(FailureInfo::ApplicationFailureInfo(
390 ApplicationFailureInfo {
391 non_retryable,
392 ..Default::default()
393 },
394 )),
395 ..ae.chain()
396 .rfold(None, |cause, e| {
397 Some(Self {
398 message: e.to_string(),
399 cause: cause.map(Box::new),
400 ..Default::default()
401 })
402 })
403 .unwrap_or_default()
404 }
405 }
406
407 pub fn timeout(timeout_type: TimeoutType) -> Self {
408 Self {
409 message: "Activity timed out".to_string(),
410 cause: Some(Box::new(Failure {
411 message: "Activity timed out".to_string(),
412 failure_info: Some(FailureInfo::TimeoutFailureInfo(TimeoutFailureInfo {
413 timeout_type: timeout_type.into(),
414 ..Default::default()
415 })),
416 ..Default::default()
417 })),
418 failure_info: Some(FailureInfo::ActivityFailureInfo(
419 ActivityFailureInfo::default(),
420 )),
421 ..Default::default()
422 }
423 }
424
425 pub fn maybe_application_failure(&self) -> Option<&ApplicationFailureInfo> {
427 if let Failure {
428 failure_info: Some(FailureInfo::ApplicationFailureInfo(f)),
429 ..
430 } = self
431 {
432 Some(f)
433 } else {
434 None
435 }
436 }
437
438 pub fn is_benign_application_failure(&self) -> bool {
440 self.maybe_application_failure()
441 .is_some_and(|app_info| app_info.category() == ApplicationErrorCategory::Benign)
442 }
443 }
444
445 impl Display for Failure {
446 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
447 write!(f, "Failure({}, ", self.message)?;
448 match self.failure_info.as_ref() {
449 None => write!(f, "missing info")?,
450 Some(FailureInfo::TimeoutFailureInfo(v)) => {
451 write!(f, "Timeout: {:?}", v.timeout_type())?;
452 }
453 Some(FailureInfo::ApplicationFailureInfo(v)) => {
454 write!(f, "Application Failure: {}", v.r#type)?;
455 }
456 Some(FailureInfo::CanceledFailureInfo(_)) => {
457 write!(f, "Cancelled")?;
458 }
459 Some(FailureInfo::TerminatedFailureInfo(_)) => {
460 write!(f, "Terminated")?;
461 }
462 Some(FailureInfo::ServerFailureInfo(_)) => {
463 write!(f, "Server Failure")?;
464 }
465 Some(FailureInfo::ResetWorkflowFailureInfo(_)) => {
466 write!(f, "Reset Workflow")?;
467 }
468 Some(FailureInfo::ActivityFailureInfo(v)) => {
469 write!(
470 f,
471 "Activity Failure: scheduled_event_id: {}",
472 v.scheduled_event_id
473 )?;
474 }
475 Some(FailureInfo::ChildWorkflowExecutionFailureInfo(v)) => {
476 write!(
477 f,
478 "Child Workflow: started_event_id: {}",
479 v.started_event_id
480 )?;
481 }
482 Some(FailureInfo::NexusOperationExecutionFailureInfo(v)) => {
483 write!(
484 f,
485 "Nexus Operation Failure: scheduled_event_id: {}",
486 v.scheduled_event_id
487 )?;
488 }
489 Some(FailureInfo::NexusHandlerFailureInfo(v)) => {
490 write!(f, "Nexus Handler Failure: {}", v.r#type)?;
491 }
492 }
493 write!(f, ")")
494 }
495 }
496
497 impl From<&str> for Failure {
498 fn from(v: &str) -> Self {
499 Failure::application_failure(v.to_string(), false)
500 }
501 }
502
503 impl From<String> for Failure {
504 fn from(v: String) -> Self {
505 Failure::application_failure(v, false)
506 }
507 }
508
509 impl From<anyhow::Error> for Failure {
510 fn from(ae: anyhow::Error) -> Self {
511 Failure::application_failure_from_error(ae, false)
512 }
513 }
514
515 pub trait FromPayloadsExt {
516 fn from_payloads(p: Option<Payloads>) -> Self;
517 }
518 impl<T> FromPayloadsExt for T
519 where
520 T: FromIterator<Payload>,
521 {
522 fn from_payloads(p: Option<Payloads>) -> Self {
523 match p {
524 None => std::iter::empty().collect(),
525 Some(p) => p.payloads.into_iter().collect(),
526 }
527 }
528 }
529
530 pub trait IntoPayloadsExt {
531 fn into_payloads(self) -> Option<Payloads>;
532 }
533 impl<T> IntoPayloadsExt for T
534 where
535 T: IntoIterator<Item = Payload>,
536 {
537 fn into_payloads(self) -> Option<Payloads> {
538 let mut iterd = self.into_iter().peekable();
539 if iterd.peek().is_none() {
540 None
541 } else {
542 Some(Payloads {
543 payloads: iterd.collect(),
544 })
545 }
546 }
547 }
548
549 impl From<Payload> for Payloads {
550 fn from(p: Payload) -> Self {
551 Self { payloads: vec![p] }
552 }
553 }
554
555 impl<T> From<T> for Payloads
556 where
557 T: AsRef<[u8]>,
558 {
559 fn from(v: T) -> Self {
560 Self {
561 payloads: vec![v.into()],
562 }
563 }
564 }
565
566 #[derive(thiserror::Error, Debug)]
567 pub enum PayloadDeserializeErr {
568 #[error("This deserializer does not understand this payload")]
571 DeserializerDoesNotHandle,
572 #[error("Error during deserialization: {0}")]
573 DeserializeErr(#[from] anyhow::Error),
574 }
575
576 pub trait AsJsonPayloadExt {
579 fn as_json_payload(&self) -> anyhow::Result<Payload>;
580 }
581 impl<T> AsJsonPayloadExt for T
582 where
583 T: Serialize,
584 {
585 fn as_json_payload(&self) -> anyhow::Result<Payload> {
586 let as_json = serde_json::to_string(self)?;
587 let mut metadata = HashMap::new();
588 metadata.insert(
589 ENCODING_PAYLOAD_KEY.to_string(),
590 JSON_ENCODING_VAL.as_bytes().to_vec(),
591 );
592 Ok(Payload {
593 metadata,
594 data: as_json.into_bytes(),
595 external_payloads: Default::default(),
596 })
597 }
598 }
599
600 pub trait FromJsonPayloadExt: Sized {
601 fn from_json_payload(payload: &Payload) -> Result<Self, PayloadDeserializeErr>;
602 }
603 impl<T> FromJsonPayloadExt for T
604 where
605 T: for<'de> Deserialize<'de>,
606 {
607 fn from_json_payload(payload: &Payload) -> Result<Self, PayloadDeserializeErr> {
608 if !payload.is_json_payload() {
609 return Err(PayloadDeserializeErr::DeserializerDoesNotHandle);
610 }
611 let payload_str =
612 std::str::from_utf8(&payload.data).map_err(anyhow::Error::from)?;
613 Ok(serde_json::from_str(payload_str).map_err(anyhow::Error::from)?)
614 }
615 }
616
617 #[derive(derive_more::Display, Debug)]
619 pub enum PayloadsToPayloadError {
620 MoreThanOnePayload,
621 NoPayload,
622 }
623 impl TryFrom<Payloads> for Payload {
624 type Error = PayloadsToPayloadError;
625
626 fn try_from(mut v: Payloads) -> Result<Self, Self::Error> {
627 match v.payloads.pop() {
628 None => Err(PayloadsToPayloadError::NoPayload),
629 Some(p) => {
630 if v.payloads.is_empty() {
631 Ok(p)
632 } else {
633 Err(PayloadsToPayloadError::MoreThanOnePayload)
634 }
635 }
636 }
637 }
638 }
639
640 pub(super) fn fix_retry_policy(mut retry_policy: RetryPolicy) -> RetryPolicy {
643 if retry_policy.initial_interval.is_none() {
644 retry_policy.initial_interval = Default::default();
645 }
646 retry_policy
647 }
648 }
649 #[allow(clippy::module_inception)]
650 pub mod activity_task {
651 tonic::include_proto!("coresdk.activity_task");
652 mod sdk_helpers {
653 use super::*;
654 use crate::protos::{coresdk::ActivityTaskCompletion, task_token::format_task_token};
655 use std::fmt::{Display, Formatter};
656
657 impl ActivityTask {
658 pub fn cancel_from_ids(
659 task_token: Vec<u8>,
660 reason: ActivityCancelReason,
661 details: ActivityCancellationDetails,
662 ) -> Self {
663 Self {
664 task_token,
665 variant: Some(activity_task::Variant::Cancel(Cancel {
666 reason: reason as i32,
667 details: Some(details),
668 })),
669 }
670 }
671
672 pub fn is_timeout(&self) -> bool {
674 match &self.variant {
675 Some(activity_task::Variant::Cancel(Cancel { reason, details })) => {
676 *reason == ActivityCancelReason::TimedOut as i32
677 || details.as_ref().is_some_and(|d| d.is_timed_out)
678 }
679 _ => false,
680 }
681 }
682
683 pub fn primary_reason_to_cancellation_details(
684 reason: ActivityCancelReason,
685 ) -> ActivityCancellationDetails {
686 ActivityCancellationDetails {
687 is_not_found: reason == ActivityCancelReason::NotFound,
688 is_cancelled: reason == ActivityCancelReason::Cancelled,
689 is_paused: reason == ActivityCancelReason::Paused,
690 is_timed_out: reason == ActivityCancelReason::TimedOut,
691 is_worker_shutdown: reason == ActivityCancelReason::WorkerShutdown,
692 is_reset: reason == ActivityCancelReason::Reset,
693 }
694 }
695 }
696
697 impl Display for ActivityTaskCompletion {
698 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
699 write!(
700 f,
701 "ActivityTaskCompletion(token: {}",
702 format_task_token(&self.task_token),
703 )?;
704 if let Some(r) = self.result.as_ref().and_then(|r| r.status.as_ref()) {
705 write!(f, ", {r}")?;
706 } else {
707 write!(f, ", missing result")?;
708 }
709 write!(f, ")")
710 }
711 }
712 }
713 }
714 #[allow(clippy::module_inception)]
715 pub mod activity_result {
716 tonic::include_proto!("coresdk.activity_result");
717 mod sdk_helpers {
718 use super::*;
719 use crate::protos::{
720 coresdk::activity_result::activity_resolution::Status,
721 temporal::api::{
722 common::v1::Payload,
723 enums::v1::TimeoutType,
724 failure::v1::{CanceledFailureInfo, Failure as APIFailure, failure},
725 },
726 };
727 use activity_execution_result as aer;
728 use anyhow::anyhow;
729 use std::fmt::{Display, Formatter};
730
731 impl ActivityExecutionResult {
732 pub const fn ok(result: Payload) -> Self {
733 Self {
734 status: Some(aer::Status::Completed(Success {
735 result: Some(result),
736 })),
737 }
738 }
739
740 pub fn fail(fail: APIFailure) -> Self {
741 Self {
742 status: Some(aer::Status::Failed(Failure {
743 failure: Some(fail),
744 })),
745 }
746 }
747
748 pub fn cancel(fail: APIFailure) -> Self {
749 Self {
750 status: Some(aer::Status::Cancelled(Cancellation {
751 failure: Some(fail),
752 })),
753 }
754 }
755
756 pub fn cancel_from_details(payload: Option<Payload>) -> Self {
757 Self {
758 status: Some(aer::Status::Cancelled(Cancellation::from_details(payload))),
759 }
760 }
761
762 pub const fn will_complete_async() -> Self {
763 Self {
764 status: Some(aer::Status::WillCompleteAsync(WillCompleteAsync {})),
765 }
766 }
767
768 pub fn is_cancelled(&self) -> bool {
769 matches!(self.status, Some(aer::Status::Cancelled(_)))
770 }
771 }
772
773 impl Display for aer::Status {
774 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
775 write!(f, "ActivityExecutionResult(")?;
776 match self {
777 aer::Status::Completed(v) => {
778 write!(f, "{v})")
779 }
780 aer::Status::Failed(v) => {
781 write!(f, "{v})")
782 }
783 aer::Status::Cancelled(v) => {
784 write!(f, "{v})")
785 }
786 aer::Status::WillCompleteAsync(_) => {
787 write!(f, "Will complete async)")
788 }
789 }
790 }
791 }
792
793 impl Display for Success {
794 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
795 write!(f, "Success(")?;
796 if let Some(ref v) = self.result {
797 write!(f, "{v}")?;
798 }
799 write!(f, ")")
800 }
801 }
802
803 impl Display for Failure {
804 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
805 write!(f, "Failure(")?;
806 if let Some(ref v) = self.failure {
807 write!(f, "{v}")?;
808 }
809 write!(f, ")")
810 }
811 }
812
813 impl Display for Cancellation {
814 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
815 write!(f, "Cancellation(")?;
816 if let Some(ref v) = self.failure {
817 write!(f, "{v}")?;
818 }
819 write!(f, ")")
820 }
821 }
822
823 impl From<Result<Payload, APIFailure>> for ActivityExecutionResult {
824 fn from(r: Result<Payload, APIFailure>) -> Self {
825 Self {
826 status: match r {
827 Ok(p) => Some(aer::Status::Completed(Success { result: Some(p) })),
828 Err(f) => Some(aer::Status::Failed(Failure { failure: Some(f) })),
829 },
830 }
831 }
832 }
833
834 impl ActivityResolution {
835 pub fn success_payload_or_error(self) -> Result<Option<Payload>, anyhow::Error> {
838 let Some(status) = self.status else {
839 return Err(anyhow!("Activity completed without a status"));
840 };
841
842 match status {
843 activity_resolution::Status::Completed(success) => Ok(success.result),
844 e => Err(anyhow!("Activity was not successful: {e:?}")),
845 }
846 }
847
848 pub fn unwrap_ok_payload(self) -> Payload {
849 self.success_payload_or_error().unwrap().unwrap()
850 }
851
852 pub fn completed_ok(&self) -> bool {
853 matches!(self.status, Some(activity_resolution::Status::Completed(_)))
854 }
855
856 pub fn failed(&self) -> bool {
857 matches!(self.status, Some(activity_resolution::Status::Failed(_)))
858 }
859
860 pub fn timed_out(&self) -> Option<TimeoutType> {
861 match self.status {
862 Some(activity_resolution::Status::Failed(Failure {
863 failure: Some(ref f),
864 })) => f.is_timeout(),
865 _ => None,
866 }
867 }
868
869 pub fn cancelled(&self) -> bool {
870 matches!(self.status, Some(activity_resolution::Status::Cancelled(_)))
871 }
872
873 pub fn unwrap_failure(self) -> APIFailure {
876 match self.status.unwrap() {
877 Status::Failed(f) => f.failure.unwrap(),
878 Status::Cancelled(c) => c.failure.unwrap(),
879 _ => panic!("Actvity did not fail"),
880 }
881 }
882 }
883
884 impl Cancellation {
885 pub fn from_details(details: Option<Payload>) -> Self {
888 Cancellation {
889 failure: Some(APIFailure {
890 message: "Activity cancelled".to_string(),
891 failure_info: Some(failure::FailureInfo::CanceledFailureInfo(
892 CanceledFailureInfo {
893 details: details.map(Into::into),
894 identity: Default::default(),
895 },
896 )),
897 ..Default::default()
898 }),
899 }
900 }
901 }
902 }
903 }
904 pub mod common {
905 tonic::include_proto!("coresdk.common");
906 pub use self::sdk_helpers::*;
907 mod sdk_helpers {
908 use crate::protos::{
909 PATCHED_MARKER_DETAILS_KEY,
910 coresdk::{
911 AsJsonPayloadExt, FromJsonPayloadExt, IntoPayloadsExt,
912 external_data::{LocalActivityMarkerData, PatchedMarkerData},
913 },
914 temporal::api::common::v1::{Payload, Payloads},
915 };
916 use std::collections::HashMap;
917
918 pub fn build_has_change_marker_details(
919 patch_id: impl Into<String>,
920 deprecated: bool,
921 ) -> anyhow::Result<HashMap<String, Payloads>> {
922 let mut hm = HashMap::new();
923 let encoded = PatchedMarkerData {
924 id: patch_id.into(),
925 deprecated,
926 }
927 .as_json_payload()?;
928 hm.insert(PATCHED_MARKER_DETAILS_KEY.to_string(), encoded.into());
929 Ok(hm)
930 }
931
932 pub fn decode_change_marker_details(
933 details: &HashMap<String, Payloads>,
934 ) -> Option<(String, bool)> {
935 if let Some(cd) = details.get(PATCHED_MARKER_DETAILS_KEY) {
938 let decoded =
939 PatchedMarkerData::from_json_payload(cd.payloads.first()?).ok()?;
940 return Some((decoded.id, decoded.deprecated));
941 }
942
943 let id_entry = details.get("patch_id")?.payloads.first()?;
944 let deprecated_entry = details.get("deprecated")?.payloads.first()?;
945 let name = std::str::from_utf8(&id_entry.data).ok()?;
946 let deprecated = *deprecated_entry.data.first()? != 0;
947 Some((name.to_string(), deprecated))
948 }
949
950 pub fn build_local_activity_marker_details(
951 metadata: LocalActivityMarkerData,
952 result: Option<Payload>,
953 ) -> HashMap<String, Payloads> {
954 let mut hm = HashMap::new();
955 if let Some(jsonified) = metadata.as_json_payload().into_payloads() {
958 hm.insert("data".to_string(), jsonified);
959 }
960 if let Some(res) = result {
961 hm.insert("result".to_string(), res.into());
962 }
963 hm
964 }
965
966 pub fn extract_local_activity_marker_data(
969 details: &HashMap<String, Payloads>,
970 ) -> Option<LocalActivityMarkerData> {
971 details
972 .get("data")
973 .and_then(|p| p.payloads.first())
974 .and_then(|p| std::str::from_utf8(&p.data).ok())
975 .and_then(|s| serde_json::from_str(s).ok())
976 }
977
978 pub fn extract_local_activity_marker_details(
982 details: &mut HashMap<String, Payloads>,
983 ) -> (Option<LocalActivityMarkerData>, Option<Payload>) {
984 let data = extract_local_activity_marker_data(details);
985 let result = details.remove("result").and_then(|mut p| p.payloads.pop());
986 (data, result)
987 }
988 }
989 }
990 pub mod external_data {
991 tonic::include_proto!("coresdk.external_data");
992 mod sdk_helpers {
993 use prost_types::{Duration, Timestamp};
994 use serde::{Deserialize, Deserializer, Serialize, Serializer};
995
996 #[derive(Serialize, Deserialize)]
1000 #[serde(remote = "Timestamp")]
1001 struct TimestampDef {
1002 seconds: i64,
1003 nanos: i32,
1004 }
1005 pub(crate) mod opt_timestamp {
1006 use super::*;
1007
1008 pub(crate) fn serialize<S>(
1009 value: &Option<Timestamp>,
1010 serializer: S,
1011 ) -> Result<S::Ok, S::Error>
1012 where
1013 S: Serializer,
1014 {
1015 #[derive(Serialize)]
1016 struct Helper<'a>(#[serde(with = "TimestampDef")] &'a Timestamp);
1017
1018 value.as_ref().map(Helper).serialize(serializer)
1019 }
1020
1021 pub(crate) fn deserialize<'de, D>(
1022 deserializer: D,
1023 ) -> Result<Option<Timestamp>, D::Error>
1024 where
1025 D: Deserializer<'de>,
1026 {
1027 #[derive(Deserialize)]
1028 struct Helper(#[serde(with = "TimestampDef")] Timestamp);
1029
1030 let helper = Option::deserialize(deserializer)?;
1031 Ok(helper.map(|Helper(external)| external))
1032 }
1033 }
1034
1035 #[derive(Serialize, Deserialize)]
1037 #[serde(remote = "Duration")]
1038 struct DurationDef {
1039 seconds: i64,
1040 nanos: i32,
1041 }
1042 pub(crate) mod opt_duration {
1043 use super::*;
1044
1045 pub(crate) fn serialize<S>(
1046 value: &Option<Duration>,
1047 serializer: S,
1048 ) -> Result<S::Ok, S::Error>
1049 where
1050 S: Serializer,
1051 {
1052 #[derive(Serialize)]
1053 struct Helper<'a>(#[serde(with = "DurationDef")] &'a Duration);
1054
1055 value.as_ref().map(Helper).serialize(serializer)
1056 }
1057
1058 pub(crate) fn deserialize<'de, D>(
1059 deserializer: D,
1060 ) -> Result<Option<Duration>, D::Error>
1061 where
1062 D: Deserializer<'de>,
1063 {
1064 #[derive(Deserialize)]
1065 struct Helper(#[serde(with = "DurationDef")] Duration);
1066
1067 let helper = Option::deserialize(deserializer)?;
1068 Ok(helper.map(|Helper(external)| external))
1069 }
1070 }
1071 }
1072 }
1073 pub mod workflow_activation {
1074 tonic::include_proto!("coresdk.workflow_activation");
1075 pub use self::sdk_helpers::*;
1076 mod sdk_helpers {
1077 use super::*;
1078 use crate::protos::{
1079 coresdk::{
1080 FromPayloadsExt,
1081 activity_result::{ActivityResolution, activity_resolution},
1082 common::NamespacedWorkflowExecution,
1083 fix_retry_policy,
1084 workflow_activation::remove_from_cache::EvictionReason,
1085 },
1086 temporal::api::{
1087 enums::v1::WorkflowTaskFailedCause,
1088 history::v1::{
1089 WorkflowExecutionCancelRequestedEventAttributes,
1090 WorkflowExecutionSignaledEventAttributes,
1091 WorkflowExecutionStartedEventAttributes,
1092 },
1093 query::v1::WorkflowQuery,
1094 },
1095 };
1096 use prost_types::Timestamp;
1097 use std::fmt::{Display, Formatter};
1098
1099 pub fn create_evict_activation(
1100 run_id: String,
1101 message: String,
1102 reason: EvictionReason,
1103 ) -> WorkflowActivation {
1104 WorkflowActivation {
1105 timestamp: None,
1106 run_id,
1107 is_replaying: false,
1108 history_length: 0,
1109 jobs: vec![WorkflowActivationJob::from(
1110 workflow_activation_job::Variant::RemoveFromCache(RemoveFromCache {
1111 message,
1112 reason: reason as i32,
1113 }),
1114 )],
1115 available_internal_flags: vec![],
1116 history_size_bytes: 0,
1117 continue_as_new_suggested: false,
1118 deployment_version_for_current_task: None,
1119 last_sdk_version: String::new(),
1120 suggest_continue_as_new_reasons: vec![],
1121 target_worker_deployment_version_changed: false,
1122 }
1123 }
1124
1125 pub fn query_to_job(id: String, q: WorkflowQuery) -> QueryWorkflow {
1126 QueryWorkflow {
1127 query_id: id,
1128 query_type: q.query_type,
1129 arguments: Vec::from_payloads(q.query_args),
1130 headers: q.header.map(|h| h.into()).unwrap_or_default(),
1131 }
1132 }
1133
1134 impl WorkflowActivation {
1135 pub fn is_only_eviction(&self) -> bool {
1137 matches!(
1138 self.jobs.as_slice(),
1139 [WorkflowActivationJob {
1140 variant: Some(workflow_activation_job::Variant::RemoveFromCache(_))
1141 }]
1142 )
1143 }
1144
1145 pub fn eviction_reason(&self) -> Option<EvictionReason> {
1147 self.jobs.iter().find_map(|j| {
1148 if let Some(workflow_activation_job::Variant::RemoveFromCache(ref rj)) =
1149 j.variant
1150 {
1151 EvictionReason::try_from(rj.reason).ok()
1152 } else {
1153 None
1154 }
1155 })
1156 }
1157 }
1158
1159 impl workflow_activation_job::Variant {
1160 pub fn is_local_activity_resolution(&self) -> bool {
1161 matches!(self, workflow_activation_job::Variant::ResolveActivity(ra) if ra.is_local)
1162 }
1163 }
1164
1165 impl Display for EvictionReason {
1166 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1167 write!(f, "{self:?}")
1168 }
1169 }
1170
1171 impl From<EvictionReason> for WorkflowTaskFailedCause {
1172 fn from(value: EvictionReason) -> Self {
1173 match value {
1174 EvictionReason::Nondeterminism => {
1175 WorkflowTaskFailedCause::NonDeterministicError
1176 }
1177 _ => WorkflowTaskFailedCause::Unspecified,
1178 }
1179 }
1180 }
1181
1182 impl Display for WorkflowActivation {
1183 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1184 write!(f, "WorkflowActivation(")?;
1185 write!(f, "run_id: {}, ", self.run_id)?;
1186 write!(f, "is_replaying: {}, ", self.is_replaying)?;
1187 write!(
1188 f,
1189 "jobs: {})",
1190 self.jobs
1191 .iter()
1192 .map(ToString::to_string)
1193 .collect::<Vec<_>>()
1194 .as_slice()
1195 .join(", ")
1196 )
1197 }
1198 }
1199
1200 impl Display for WorkflowActivationJob {
1201 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1202 match &self.variant {
1203 None => write!(f, "empty"),
1204 Some(v) => write!(f, "{v}"),
1205 }
1206 }
1207 }
1208
1209 impl Display for workflow_activation_job::Variant {
1210 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1211 match self {
1212 workflow_activation_job::Variant::InitializeWorkflow(_) => {
1213 write!(f, "InitializeWorkflow")
1214 }
1215 workflow_activation_job::Variant::FireTimer(t) => {
1216 write!(f, "FireTimer({})", t.seq)
1217 }
1218 workflow_activation_job::Variant::UpdateRandomSeed(_) => {
1219 write!(f, "UpdateRandomSeed")
1220 }
1221 workflow_activation_job::Variant::QueryWorkflow(_) => {
1222 write!(f, "QueryWorkflow")
1223 }
1224 workflow_activation_job::Variant::CancelWorkflow(_) => {
1225 write!(f, "CancelWorkflow")
1226 }
1227 workflow_activation_job::Variant::SignalWorkflow(_) => {
1228 write!(f, "SignalWorkflow")
1229 }
1230 workflow_activation_job::Variant::ResolveActivity(r) => {
1231 write!(
1232 f,
1233 "ResolveActivity({}, {})",
1234 r.seq,
1235 r.result
1236 .as_ref()
1237 .unwrap_or(&ActivityResolution { status: None })
1238 )
1239 }
1240 workflow_activation_job::Variant::NotifyHasPatch(_) => {
1241 write!(f, "NotifyHasPatch")
1242 }
1243 workflow_activation_job::Variant::ResolveChildWorkflowExecutionStart(_) => {
1244 write!(f, "ResolveChildWorkflowExecutionStart")
1245 }
1246 workflow_activation_job::Variant::ResolveChildWorkflowExecution(_) => {
1247 write!(f, "ResolveChildWorkflowExecution")
1248 }
1249 workflow_activation_job::Variant::ResolveSignalExternalWorkflow(_) => {
1250 write!(f, "ResolveSignalExternalWorkflow")
1251 }
1252 workflow_activation_job::Variant::RemoveFromCache(_) => {
1253 write!(f, "RemoveFromCache")
1254 }
1255 workflow_activation_job::Variant::ResolveRequestCancelExternalWorkflow(
1256 _,
1257 ) => {
1258 write!(f, "ResolveRequestCancelExternalWorkflow")
1259 }
1260 workflow_activation_job::Variant::DoUpdate(u) => {
1261 write!(f, "DoUpdate({})", u.id)
1262 }
1263 workflow_activation_job::Variant::ResolveNexusOperationStart(_) => {
1264 write!(f, "ResolveNexusOperationStart")
1265 }
1266 workflow_activation_job::Variant::ResolveNexusOperation(_) => {
1267 write!(f, "ResolveNexusOperation")
1268 }
1269 }
1270 }
1271 }
1272
1273 impl Display for ActivityResolution {
1274 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1275 match self.status {
1276 None => {
1277 write!(f, "None")
1278 }
1279 Some(activity_resolution::Status::Failed(_)) => {
1280 write!(f, "Failed")
1281 }
1282 Some(activity_resolution::Status::Completed(_)) => {
1283 write!(f, "Completed")
1284 }
1285 Some(activity_resolution::Status::Cancelled(_)) => {
1286 write!(f, "Cancelled")
1287 }
1288 Some(activity_resolution::Status::Backoff(_)) => {
1289 write!(f, "Backoff")
1290 }
1291 }
1292 }
1293 }
1294
1295 impl Display for QueryWorkflow {
1296 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1297 write!(
1298 f,
1299 "QueryWorkflow(id: {}, type: {})",
1300 self.query_id, self.query_type
1301 )
1302 }
1303 }
1304
1305 impl From<WorkflowExecutionSignaledEventAttributes> for SignalWorkflow {
1306 fn from(a: WorkflowExecutionSignaledEventAttributes) -> Self {
1307 Self {
1308 signal_name: a.signal_name,
1309 input: Vec::from_payloads(a.input),
1310 identity: a.identity,
1311 headers: a.header.map(Into::into).unwrap_or_default(),
1312 }
1313 }
1314 }
1315
1316 impl From<WorkflowExecutionCancelRequestedEventAttributes> for CancelWorkflow {
1317 fn from(a: WorkflowExecutionCancelRequestedEventAttributes) -> Self {
1318 Self { reason: a.cause }
1319 }
1320 }
1321
1322 pub fn start_workflow_from_attribs(
1324 attrs: WorkflowExecutionStartedEventAttributes,
1325 workflow_id: String,
1326 randomness_seed: u64,
1327 start_time: Timestamp,
1328 ) -> InitializeWorkflow {
1329 InitializeWorkflow {
1330 workflow_type: attrs.workflow_type.map(|wt| wt.name).unwrap_or_default(),
1331 workflow_id,
1332 arguments: Vec::from_payloads(attrs.input),
1333 randomness_seed,
1334 headers: attrs.header.unwrap_or_default().fields,
1335 identity: attrs.identity,
1336 parent_workflow_info: attrs.parent_workflow_execution.map(|pe| {
1337 NamespacedWorkflowExecution {
1338 namespace: attrs.parent_workflow_namespace,
1339 run_id: pe.run_id,
1340 workflow_id: pe.workflow_id,
1341 }
1342 }),
1343 workflow_execution_timeout: attrs.workflow_execution_timeout,
1344 workflow_run_timeout: attrs.workflow_run_timeout,
1345 workflow_task_timeout: attrs.workflow_task_timeout,
1346 continued_from_execution_run_id: attrs.continued_execution_run_id,
1347 continued_initiator: attrs.initiator,
1348 continued_failure: attrs.continued_failure,
1349 last_completion_result: attrs.last_completion_result,
1350 first_execution_run_id: attrs.first_execution_run_id,
1351 retry_policy: attrs.retry_policy.map(fix_retry_policy),
1352 attempt: attrs.attempt,
1353 cron_schedule: attrs.cron_schedule,
1354 workflow_execution_expiration_time: attrs.workflow_execution_expiration_time,
1355 cron_schedule_to_schedule_interval: attrs.first_workflow_task_backoff,
1356 memo: attrs.memo,
1357 search_attributes: attrs.search_attributes,
1358 start_time: Some(start_time),
1359 root_workflow: attrs.root_workflow_execution,
1360 priority: attrs.priority,
1361 }
1362 }
1363 }
1364 }
1365 pub mod workflow_completion {
1366 tonic::include_proto!("coresdk.workflow_completion");
1367 mod sdk_helpers {
1368 use super::*;
1369 use crate::protos::temporal::api::{enums::v1::WorkflowTaskFailedCause, failure};
1370
1371 impl workflow_activation_completion::Status {
1372 pub const fn is_success(&self) -> bool {
1373 match &self {
1374 Self::Successful(_) => true,
1375 Self::Failed(_) => false,
1376 }
1377 }
1378 }
1379
1380 impl From<failure::v1::Failure> for Failure {
1381 fn from(f: failure::v1::Failure) -> Self {
1382 Failure {
1383 failure: Some(f),
1384 force_cause: WorkflowTaskFailedCause::Unspecified as i32,
1385 }
1386 }
1387 }
1388 }
1389 }
1390 pub mod child_workflow {
1391 tonic::include_proto!("coresdk.child_workflow");
1392 }
1393 pub mod nexus {
1394 tonic::include_proto!("coresdk.nexus");
1395 pub use self::sdk_helpers::*;
1396 mod sdk_helpers {
1397 use super::*;
1398 use crate::protos::temporal::api::workflowservice::v1::PollNexusTaskQueueResponse;
1399 use std::fmt::{Display, Formatter};
1400
1401 impl NexusTask {
1402 pub fn unwrap_task(self) -> PollNexusTaskQueueResponse {
1404 if let Some(nexus_task::Variant::Task(t)) = self.variant {
1405 return t;
1406 }
1407 panic!("Nexus task did not contain a server task");
1408 }
1409
1410 pub fn task_token(&self) -> &[u8] {
1412 match &self.variant {
1413 Some(nexus_task::Variant::Task(t)) => t.task_token.as_slice(),
1414 Some(nexus_task::Variant::CancelTask(c)) => c.task_token.as_slice(),
1415 None => panic!("Nexus task did not contain a task token"),
1416 }
1417 }
1418 }
1419
1420 impl Display for nexus_task_completion::Status {
1421 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1422 write!(f, "NexusTaskCompletion(")?;
1423 match self {
1424 nexus_task_completion::Status::Completed(c) => {
1425 write!(f, "{c}")
1426 }
1427 nexus_task_completion::Status::AckCancel(_) => {
1428 write!(f, "AckCancel")
1429 }
1430 #[allow(deprecated)]
1431 nexus_task_completion::Status::Error(error) => {
1432 write!(f, "Error({error:?})")
1433 }
1434 nexus_task_completion::Status::Failure(failure) => {
1435 write!(f, "{failure}")
1436 }
1437 }?;
1438 write!(f, ")")
1439 }
1440 }
1441
1442 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
1443 pub enum NexusOperationErrorState {
1444 Failed,
1445 Canceled,
1446 }
1447
1448 impl Display for NexusOperationErrorState {
1449 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1450 match self {
1451 Self::Failed => write!(f, "failed"),
1452 Self::Canceled => write!(f, "canceled"),
1453 }
1454 }
1455 }
1456 }
1457 }
1458 pub mod workflow_commands {
1459 tonic::include_proto!("coresdk.workflow_commands");
1460 mod sdk_helpers {
1461 use super::*;
1462 use crate::protos::temporal::api::{common::v1::Payloads, enums::v1::QueryResultType};
1463 use std::fmt::{Display, Formatter};
1464
1465 impl Display for WorkflowCommand {
1466 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1467 match &self.variant {
1468 None => write!(f, "Empty"),
1469 Some(v) => write!(f, "{v}"),
1470 }
1471 }
1472 }
1473
1474 impl Display for StartTimer {
1475 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1476 write!(f, "StartTimer({})", self.seq)
1477 }
1478 }
1479
1480 impl Display for ScheduleActivity {
1481 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1482 write!(f, "ScheduleActivity({}, {})", self.seq, self.activity_type)
1483 }
1484 }
1485
1486 impl Display for ScheduleLocalActivity {
1487 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1488 write!(
1489 f,
1490 "ScheduleLocalActivity({}, {})",
1491 self.seq, self.activity_type
1492 )
1493 }
1494 }
1495
1496 impl Display for QueryResult {
1497 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1498 write!(f, "RespondToQuery({})", self.query_id)
1499 }
1500 }
1501
1502 impl Display for RequestCancelActivity {
1503 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1504 write!(f, "RequestCancelActivity({})", self.seq)
1505 }
1506 }
1507
1508 impl Display for RequestCancelLocalActivity {
1509 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1510 write!(f, "RequestCancelLocalActivity({})", self.seq)
1511 }
1512 }
1513
1514 impl Display for CancelTimer {
1515 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1516 write!(f, "CancelTimer({})", self.seq)
1517 }
1518 }
1519
1520 impl Display for CompleteWorkflowExecution {
1521 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1522 write!(f, "CompleteWorkflowExecution")
1523 }
1524 }
1525
1526 impl Display for FailWorkflowExecution {
1527 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1528 write!(f, "FailWorkflowExecution")
1529 }
1530 }
1531
1532 impl Display for ContinueAsNewWorkflowExecution {
1533 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1534 write!(f, "ContinueAsNewWorkflowExecution")
1535 }
1536 }
1537
1538 impl Display for CancelWorkflowExecution {
1539 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1540 write!(f, "CancelWorkflowExecution")
1541 }
1542 }
1543
1544 impl Display for SetPatchMarker {
1545 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1546 write!(f, "SetPatchMarker({})", self.patch_id)
1547 }
1548 }
1549
1550 impl Display for StartChildWorkflowExecution {
1551 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1552 write!(
1553 f,
1554 "StartChildWorkflowExecution({}, {})",
1555 self.seq, self.workflow_type
1556 )
1557 }
1558 }
1559
1560 impl Display for RequestCancelExternalWorkflowExecution {
1561 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1562 write!(f, "RequestCancelExternalWorkflowExecution({})", self.seq)
1563 }
1564 }
1565
1566 impl Display for UpsertWorkflowSearchAttributes {
1567 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1568 let keys: Vec<_> = self
1569 .search_attributes
1570 .as_ref()
1571 .map(|sa| sa.indexed_fields.keys().collect())
1572 .unwrap_or_default();
1573 write!(f, "UpsertWorkflowSearchAttributes({:?})", keys)
1574 }
1575 }
1576
1577 impl Display for SignalExternalWorkflowExecution {
1578 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1579 write!(f, "SignalExternalWorkflowExecution({})", self.seq)
1580 }
1581 }
1582
1583 impl Display for CancelSignalWorkflow {
1584 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1585 write!(f, "CancelSignalWorkflow({})", self.seq)
1586 }
1587 }
1588
1589 impl Display for CancelChildWorkflowExecution {
1590 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1591 write!(
1592 f,
1593 "CancelChildWorkflowExecution({})",
1594 self.child_workflow_seq
1595 )
1596 }
1597 }
1598
1599 impl Display for ModifyWorkflowProperties {
1600 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1601 write!(
1602 f,
1603 "ModifyWorkflowProperties(upserted memo keys: {:?})",
1604 self.upserted_memo.as_ref().map(|m| m.fields.keys())
1605 )
1606 }
1607 }
1608
1609 impl Display for UpdateResponse {
1610 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1611 write!(
1612 f,
1613 "UpdateResponse(protocol_instance_id: {}, response: {:?})",
1614 self.protocol_instance_id, self.response
1615 )
1616 }
1617 }
1618
1619 impl Display for ScheduleNexusOperation {
1620 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1621 write!(f, "ScheduleNexusOperation({})", self.seq)
1622 }
1623 }
1624
1625 impl Display for RequestCancelNexusOperation {
1626 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1627 write!(f, "RequestCancelNexusOperation({})", self.seq)
1628 }
1629 }
1630
1631 impl QueryResult {
1632 pub fn into_components(
1634 self,
1635 ) -> (String, QueryResultType, Option<Payloads>, String) {
1636 match self {
1637 QueryResult {
1638 variant: Some(query_result::Variant::Succeeded(qs)),
1639 query_id,
1640 } => (
1641 query_id,
1642 QueryResultType::Answered,
1643 qs.response.map(Into::into),
1644 "".to_string(),
1645 ),
1646 QueryResult {
1647 variant: Some(query_result::Variant::Failed(err)),
1648 query_id,
1649 } => (query_id, QueryResultType::Failed, None, err.message),
1650 QueryResult {
1651 variant: None,
1652 query_id,
1653 } => (
1654 query_id,
1655 QueryResultType::Failed,
1656 None,
1657 "Query response was empty".to_string(),
1658 ),
1659 }
1660 }
1661 }
1662 }
1663 }
1664}
1665
1666#[allow(
1668 clippy::all,
1669 missing_docs,
1670 rustdoc::broken_intra_doc_links,
1671 rustdoc::bare_urls
1672)]
1673pub mod temporal {
1675 pub mod api {
1676 pub mod activity {
1677 pub mod v1 {
1678 tonic::include_proto!("temporal.api.activity.v1");
1679 }
1680 }
1681 pub mod batch {
1682 pub mod v1 {
1683 tonic::include_proto!("temporal.api.batch.v1");
1684 }
1685 }
1686 pub mod callback {
1687 pub mod v1 {
1688 tonic::include_proto!("temporal.api.callback.v1");
1689 }
1690 }
1691 pub mod command {
1692 pub mod v1 {
1693 tonic::include_proto!("temporal.api.command.v1");
1694 pub use self::sdk_helpers::*;
1695 mod sdk_helpers {
1696 use super::*;
1697 use crate::protos::{
1698 coresdk::{IntoPayloadsExt, workflow_commands},
1699 temporal::api::{
1700 common::v1::{ActivityType, WorkflowType},
1701 enums::v1::CommandType,
1702 },
1703 };
1704 use command::Attributes;
1705 use std::fmt::{Display, Formatter};
1706
1707 impl From<command::Attributes> for Command {
1708 fn from(c: command::Attributes) -> Self {
1709 match c {
1710 a @ Attributes::StartTimerCommandAttributes(_) => Self {
1711 command_type: CommandType::StartTimer as i32,
1712 attributes: Some(a),
1713 user_metadata: Default::default(),
1714 event_group_markers: Default::default(),
1715 },
1716 a @ Attributes::CancelTimerCommandAttributes(_) => Self {
1717 command_type: CommandType::CancelTimer as i32,
1718 attributes: Some(a),
1719 user_metadata: Default::default(),
1720 event_group_markers: Default::default(),
1721 },
1722 a @ Attributes::CompleteWorkflowExecutionCommandAttributes(_) => {
1723 Self {
1724 command_type: CommandType::CompleteWorkflowExecution as i32,
1725 attributes: Some(a),
1726 user_metadata: Default::default(),
1727 event_group_markers: Default::default(),
1728 }
1729 }
1730 a @ Attributes::FailWorkflowExecutionCommandAttributes(_) => Self {
1731 command_type: CommandType::FailWorkflowExecution as i32,
1732 attributes: Some(a),
1733 user_metadata: Default::default(),
1734 event_group_markers: Default::default(),
1735 },
1736 a @ Attributes::ScheduleActivityTaskCommandAttributes(_) => Self {
1737 command_type: CommandType::ScheduleActivityTask as i32,
1738 attributes: Some(a),
1739 user_metadata: Default::default(),
1740 event_group_markers: Default::default(),
1741 },
1742 a @ Attributes::RequestCancelActivityTaskCommandAttributes(_) => {
1743 Self {
1744 command_type: CommandType::RequestCancelActivityTask as i32,
1745 attributes: Some(a),
1746 user_metadata: Default::default(),
1747 event_group_markers: Default::default(),
1748 }
1749 }
1750 a
1751 @ Attributes::ContinueAsNewWorkflowExecutionCommandAttributes(
1752 _,
1753 ) => Self {
1754 command_type: CommandType::ContinueAsNewWorkflowExecution
1755 as i32,
1756 attributes: Some(a),
1757 user_metadata: Default::default(),
1758 event_group_markers: Default::default(),
1759 },
1760 a @ Attributes::CancelWorkflowExecutionCommandAttributes(_) => {
1761 Self {
1762 command_type: CommandType::CancelWorkflowExecution as i32,
1763 attributes: Some(a),
1764 user_metadata: Default::default(),
1765 event_group_markers: Default::default(),
1766 }
1767 }
1768 a @ Attributes::RecordMarkerCommandAttributes(_) => Self {
1769 command_type: CommandType::RecordMarker as i32,
1770 attributes: Some(a),
1771 user_metadata: Default::default(),
1772 event_group_markers: Default::default(),
1773 },
1774 a @ Attributes::ProtocolMessageCommandAttributes(_) => Self {
1775 command_type: CommandType::ProtocolMessage as i32,
1776 attributes: Some(a),
1777 user_metadata: Default::default(),
1778 event_group_markers: Default::default(),
1779 },
1780 a @ Attributes::RequestCancelNexusOperationCommandAttributes(_) => {
1781 Self {
1782 command_type: CommandType::RequestCancelNexusOperation
1783 as i32,
1784 attributes: Some(a),
1785 user_metadata: Default::default(),
1786 event_group_markers: Default::default(),
1787 }
1788 }
1789 _ => unimplemented!(),
1790 }
1791 }
1792 }
1793
1794 impl Display for Command {
1795 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1796 let ct = CommandType::try_from(self.command_type)
1797 .unwrap_or(CommandType::Unspecified);
1798 write!(f, "{:?}", ct)
1799 }
1800 }
1801
1802 pub trait CommandAttributesExt {
1803 fn as_type(&self) -> CommandType;
1804 }
1805
1806 impl CommandAttributesExt for command::Attributes {
1807 fn as_type(&self) -> CommandType {
1808 match self {
1809 Attributes::ScheduleActivityTaskCommandAttributes(_) => {
1810 CommandType::ScheduleActivityTask
1811 }
1812 Attributes::StartTimerCommandAttributes(_) => CommandType::StartTimer,
1813 Attributes::CompleteWorkflowExecutionCommandAttributes(_) => {
1814 CommandType::CompleteWorkflowExecution
1815 }
1816 Attributes::FailWorkflowExecutionCommandAttributes(_) => {
1817 CommandType::FailWorkflowExecution
1818 }
1819 Attributes::RequestCancelActivityTaskCommandAttributes(_) => {
1820 CommandType::RequestCancelActivityTask
1821 }
1822 Attributes::CancelTimerCommandAttributes(_) => CommandType::CancelTimer,
1823 Attributes::CancelWorkflowExecutionCommandAttributes(_) => {
1824 CommandType::CancelWorkflowExecution
1825 }
1826 Attributes::RequestCancelExternalWorkflowExecutionCommandAttributes(
1827 _,
1828 ) => CommandType::RequestCancelExternalWorkflowExecution,
1829 Attributes::RecordMarkerCommandAttributes(_) => {
1830 CommandType::RecordMarker
1831 }
1832 Attributes::ContinueAsNewWorkflowExecutionCommandAttributes(_) => {
1833 CommandType::ContinueAsNewWorkflowExecution
1834 }
1835 Attributes::StartChildWorkflowExecutionCommandAttributes(_) => {
1836 CommandType::StartChildWorkflowExecution
1837 }
1838 Attributes::SignalExternalWorkflowExecutionCommandAttributes(_) => {
1839 CommandType::SignalExternalWorkflowExecution
1840 }
1841 Attributes::UpsertWorkflowSearchAttributesCommandAttributes(_) => {
1842 CommandType::UpsertWorkflowSearchAttributes
1843 }
1844 Attributes::ProtocolMessageCommandAttributes(_) => {
1845 CommandType::ProtocolMessage
1846 }
1847 Attributes::ModifyWorkflowPropertiesCommandAttributes(_) => {
1848 CommandType::ModifyWorkflowProperties
1849 }
1850 Attributes::ScheduleNexusOperationCommandAttributes(_) => {
1851 CommandType::ScheduleNexusOperation
1852 }
1853 Attributes::RequestCancelNexusOperationCommandAttributes(_) => {
1854 CommandType::RequestCancelNexusOperation
1855 }
1856 }
1857 }
1858 }
1859
1860 impl Display for command::Attributes {
1861 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1862 write!(f, "{:?}", self.as_type())
1863 }
1864 }
1865
1866 impl From<workflow_commands::StartTimer> for command::Attributes {
1867 fn from(s: workflow_commands::StartTimer) -> Self {
1868 Self::StartTimerCommandAttributes(StartTimerCommandAttributes {
1869 timer_id: s.seq.to_string(),
1870 start_to_fire_timeout: s.start_to_fire_timeout,
1871 })
1872 }
1873 }
1874
1875 impl From<workflow_commands::UpsertWorkflowSearchAttributes> for command::Attributes {
1876 fn from(s: workflow_commands::UpsertWorkflowSearchAttributes) -> Self {
1877 Self::UpsertWorkflowSearchAttributesCommandAttributes(
1878 UpsertWorkflowSearchAttributesCommandAttributes {
1879 search_attributes: s.search_attributes,
1880 },
1881 )
1882 }
1883 }
1884
1885 impl From<workflow_commands::ModifyWorkflowProperties> for command::Attributes {
1886 fn from(s: workflow_commands::ModifyWorkflowProperties) -> Self {
1887 Self::ModifyWorkflowPropertiesCommandAttributes(
1888 ModifyWorkflowPropertiesCommandAttributes {
1889 upserted_memo: s.upserted_memo.map(Into::into),
1890 },
1891 )
1892 }
1893 }
1894
1895 impl From<workflow_commands::CancelTimer> for command::Attributes {
1896 fn from(s: workflow_commands::CancelTimer) -> Self {
1897 Self::CancelTimerCommandAttributes(CancelTimerCommandAttributes {
1898 timer_id: s.seq.to_string(),
1899 })
1900 }
1901 }
1902
1903 pub fn schedule_activity_cmd_to_api(
1904 s: workflow_commands::ScheduleActivity,
1905 use_workflow_build_id: bool,
1906 ) -> command::Attributes {
1907 command::Attributes::ScheduleActivityTaskCommandAttributes(
1908 ScheduleActivityTaskCommandAttributes {
1909 activity_id: s.activity_id,
1910 activity_type: Some(ActivityType {
1911 name: s.activity_type,
1912 }),
1913 task_queue: Some(s.task_queue.into()),
1914 header: Some(s.headers.into()),
1915 input: s.arguments.into_payloads(),
1916 schedule_to_close_timeout: s.schedule_to_close_timeout,
1917 schedule_to_start_timeout: s.schedule_to_start_timeout,
1918 start_to_close_timeout: s.start_to_close_timeout,
1919 heartbeat_timeout: s.heartbeat_timeout,
1920 retry_policy: s.retry_policy.map(Into::into),
1921 request_eager_execution: !s.do_not_eagerly_execute,
1922 use_workflow_build_id,
1923 priority: s.priority,
1924 },
1925 )
1926 }
1927
1928 #[allow(deprecated)]
1929 pub fn start_child_workflow_cmd_to_api(
1930 s: workflow_commands::StartChildWorkflowExecution,
1931 inherit_build_id: bool,
1932 ) -> command::Attributes {
1933 command::Attributes::StartChildWorkflowExecutionCommandAttributes(
1934 StartChildWorkflowExecutionCommandAttributes {
1935 workflow_id: s.workflow_id,
1936 workflow_type: Some(WorkflowType {
1937 name: s.workflow_type,
1938 }),
1939 control: "".into(),
1940 namespace: s.namespace,
1941 task_queue: Some(s.task_queue.into()),
1942 header: Some(s.headers.into()),
1943 memo: Some(s.memo.into()),
1944 search_attributes: s.search_attributes,
1945 input: s.input.into_payloads(),
1946 workflow_id_reuse_policy: s.workflow_id_reuse_policy,
1947 workflow_execution_timeout: s.workflow_execution_timeout,
1948 workflow_run_timeout: s.workflow_run_timeout,
1949 workflow_task_timeout: s.workflow_task_timeout,
1950 retry_policy: s.retry_policy.map(Into::into),
1951 cron_schedule: s.cron_schedule.clone(),
1952 parent_close_policy: s.parent_close_policy,
1953 inherit_build_id,
1954 priority: s.priority,
1955 versioning_override: None,
1956 },
1957 )
1958 }
1959
1960 impl From<workflow_commands::CompleteWorkflowExecution> for command::Attributes {
1961 fn from(c: workflow_commands::CompleteWorkflowExecution) -> Self {
1962 Self::CompleteWorkflowExecutionCommandAttributes(
1963 CompleteWorkflowExecutionCommandAttributes {
1964 result: c.result.map(Into::into),
1965 },
1966 )
1967 }
1968 }
1969
1970 impl From<workflow_commands::FailWorkflowExecution> for command::Attributes {
1971 fn from(c: workflow_commands::FailWorkflowExecution) -> Self {
1972 Self::FailWorkflowExecutionCommandAttributes(
1973 FailWorkflowExecutionCommandAttributes {
1974 failure: c.failure.map(Into::into),
1975 },
1976 )
1977 }
1978 }
1979
1980 #[allow(deprecated)]
1981 pub fn continue_as_new_cmd_to_api(
1982 c: workflow_commands::ContinueAsNewWorkflowExecution,
1983 inherit_build_id: bool,
1984 ) -> command::Attributes {
1985 command::Attributes::ContinueAsNewWorkflowExecutionCommandAttributes(
1986 ContinueAsNewWorkflowExecutionCommandAttributes {
1987 workflow_type: Some(c.workflow_type.into()),
1988 task_queue: Some(c.task_queue.into()),
1989 input: c.arguments.into_payloads(),
1990 workflow_run_timeout: c.workflow_run_timeout,
1991 workflow_task_timeout: c.workflow_task_timeout,
1992 memo: if c.memo.is_empty() {
1993 None
1994 } else {
1995 Some(c.memo.into())
1996 },
1997 header: if c.headers.is_empty() {
1998 None
1999 } else {
2000 Some(c.headers.into())
2001 },
2002 retry_policy: c.retry_policy,
2003 search_attributes: c.search_attributes,
2004 backoff_start_interval: c.backoff_start_interval,
2005 inherit_build_id,
2006 initial_versioning_behavior: c.initial_versioning_behavior,
2007 ..Default::default()
2008 },
2009 )
2010 }
2011
2012 impl From<workflow_commands::CancelWorkflowExecution> for command::Attributes {
2013 fn from(_c: workflow_commands::CancelWorkflowExecution) -> Self {
2014 Self::CancelWorkflowExecutionCommandAttributes(
2015 CancelWorkflowExecutionCommandAttributes { details: None },
2016 )
2017 }
2018 }
2019
2020 impl From<workflow_commands::ScheduleNexusOperation> for command::Attributes {
2021 fn from(c: workflow_commands::ScheduleNexusOperation) -> Self {
2022 Self::ScheduleNexusOperationCommandAttributes(
2023 ScheduleNexusOperationCommandAttributes {
2024 endpoint: c.endpoint,
2025 service: c.service,
2026 operation: c.operation,
2027 input: c.input,
2028 schedule_to_close_timeout: c.schedule_to_close_timeout,
2029 schedule_to_start_timeout: c.schedule_to_start_timeout,
2030 start_to_close_timeout: c.start_to_close_timeout,
2031 nexus_header: c.nexus_header,
2032 },
2033 )
2034 }
2035 }
2036 }
2037 }
2038 }
2039 #[allow(rustdoc::invalid_html_tags)]
2040 pub mod cloud {
2041 pub mod account {
2042 pub mod v1 {
2043 tonic::include_proto!("temporal.api.cloud.account.v1");
2044 }
2045 }
2046 pub mod auditlog {
2047 pub mod v1 {
2048 tonic::include_proto!("temporal.api.cloud.auditlog.v1");
2049 }
2050 }
2051 pub mod billing {
2052 pub mod v1 {
2053 tonic::include_proto!("temporal.api.cloud.billing.v1");
2054 }
2055 }
2056 pub mod cloudservice {
2057 pub mod v1 {
2058 tonic::include_proto!("temporal.api.cloud.cloudservice.v1");
2059 }
2060 }
2061 pub mod connectivityrule {
2062 pub mod v1 {
2063 tonic::include_proto!("temporal.api.cloud.connectivityrule.v1");
2064 }
2065 }
2066 pub mod identity {
2067 pub mod v1 {
2068 tonic::include_proto!("temporal.api.cloud.identity.v1");
2069 }
2070 }
2071 pub mod namespace {
2072 pub mod v1 {
2073 tonic::include_proto!("temporal.api.cloud.namespace.v1");
2074 }
2075 }
2076 pub mod nexus {
2077 pub mod v1 {
2078 tonic::include_proto!("temporal.api.cloud.nexus.v1");
2079 }
2080 }
2081 pub mod operation {
2082 pub mod v1 {
2083 tonic::include_proto!("temporal.api.cloud.operation.v1");
2084 }
2085 }
2086 pub mod region {
2087 pub mod v1 {
2088 tonic::include_proto!("temporal.api.cloud.region.v1");
2089 }
2090 }
2091 pub mod resource {
2092 pub mod v1 {
2093 tonic::include_proto!("temporal.api.cloud.resource.v1");
2094 }
2095 }
2096 pub mod sink {
2097 pub mod v1 {
2098 tonic::include_proto!("temporal.api.cloud.sink.v1");
2099 }
2100 }
2101 pub mod usage {
2102 pub mod v1 {
2103 tonic::include_proto!("temporal.api.cloud.usage.v1");
2104 }
2105 }
2106 }
2107 pub mod common {
2108 pub mod v1 {
2109 include_proto_with_serde!("temporal.api.common.v1");
2110 mod sdk_helpers {
2111 use super::*;
2112 use crate::protos::{ENCODING_PAYLOAD_KEY, JSON_ENCODING_VAL};
2113 use base64::{Engine, prelude::BASE64_STANDARD};
2114 use std::{
2115 collections::HashMap,
2116 fmt::{Display, Formatter},
2117 };
2118
2119 impl<T> From<T> for Payload
2120 where
2121 T: AsRef<[u8]>,
2122 {
2123 fn from(v: T) -> Self {
2124 let mut metadata = HashMap::new();
2127 metadata
2128 .insert(ENCODING_PAYLOAD_KEY.to_string(), b"binary/plain".to_vec());
2129 Self {
2130 metadata,
2131 data: v.as_ref().to_vec(),
2132 external_payloads: Default::default(),
2133 }
2134 }
2135 }
2136
2137 impl Payload {
2138 pub fn as_slice(&self) -> &[u8] {
2140 self.data.as_slice()
2141 }
2142
2143 pub fn is_json_payload(&self) -> bool {
2144 self.metadata
2145 .get(ENCODING_PAYLOAD_KEY)
2146 .map(|v| v.as_slice() == JSON_ENCODING_VAL.as_bytes())
2147 .unwrap_or_default()
2148 }
2149 }
2150
2151 impl std::fmt::Debug for Payload {
2152 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2153 if std::env::var("TEMPORAL_PRINT_FULL_PAYLOADS").is_err()
2154 && self.data.len() > 64
2155 {
2156 let mut windows = self.data.as_slice().windows(32);
2157 write!(
2158 f,
2159 "[{}..{}]",
2160 BASE64_STANDARD.encode(windows.next().unwrap_or_default()),
2161 BASE64_STANDARD.encode(windows.next_back().unwrap_or_default())
2162 )
2163 } else {
2164 write!(f, "[{}]", BASE64_STANDARD.encode(&self.data))
2165 }
2166 }
2167 }
2168
2169 impl Display for Payload {
2170 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2171 write!(f, "{:?}", self)
2172 }
2173 }
2174
2175 impl Display for Header {
2176 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2177 write!(f, "Header(")?;
2178 for kv in &self.fields {
2179 write!(f, "{}: ", kv.0)?;
2180 write!(f, "{}, ", kv.1)?;
2181 }
2182 write!(f, ")")
2183 }
2184 }
2185
2186 impl From<Header> for HashMap<String, Payload> {
2187 fn from(h: Header) -> Self {
2188 h.fields.into_iter().map(|(k, v)| (k, v.into())).collect()
2189 }
2190 }
2191
2192 impl From<Memo> for HashMap<String, Payload> {
2193 fn from(h: Memo) -> Self {
2194 h.fields.into_iter().map(|(k, v)| (k, v.into())).collect()
2195 }
2196 }
2197
2198 impl From<SearchAttributes> for HashMap<String, Payload> {
2199 fn from(h: SearchAttributes) -> Self {
2200 h.indexed_fields
2201 .into_iter()
2202 .map(|(k, v)| (k, v.into()))
2203 .collect()
2204 }
2205 }
2206
2207 impl From<HashMap<String, Payload>> for SearchAttributes {
2208 fn from(h: HashMap<String, Payload>) -> Self {
2209 Self {
2210 indexed_fields: h.into_iter().map(|(k, v)| (k, v.into())).collect(),
2211 }
2212 }
2213 }
2214
2215 impl From<String> for ActivityType {
2216 fn from(name: String) -> Self {
2217 Self { name }
2218 }
2219 }
2220
2221 impl From<&str> for ActivityType {
2222 fn from(name: &str) -> Self {
2223 Self {
2224 name: name.to_string(),
2225 }
2226 }
2227 }
2228
2229 impl From<ActivityType> for String {
2230 fn from(at: ActivityType) -> Self {
2231 at.name
2232 }
2233 }
2234
2235 impl From<&str> for WorkflowType {
2236 fn from(v: &str) -> Self {
2237 Self {
2238 name: v.to_string(),
2239 }
2240 }
2241 }
2242 }
2243 }
2244 }
2245 pub mod compute {
2246 pub mod v1 {
2247 tonic::include_proto!("temporal.api.compute.v1");
2248 }
2249 }
2250 pub mod deployment {
2251 pub mod v1 {
2252 tonic::include_proto!("temporal.api.deployment.v1");
2253 }
2254 }
2255 pub mod enums {
2256 pub mod v1 {
2257 include_proto_with_serde!("temporal.api.enums.v1");
2258 }
2259 }
2260 pub mod errordetails {
2261 pub mod v1 {
2262 tonic::include_proto!("temporal.api.errordetails.v1");
2263 }
2264 }
2265 pub mod failure {
2266 pub mod v1 {
2267 include_proto_with_serde!("temporal.api.failure.v1");
2268 }
2269 }
2270 pub mod filter {
2271 pub mod v1 {
2272 tonic::include_proto!("temporal.api.filter.v1");
2273 }
2274 }
2275 pub mod history {
2276 pub mod v1 {
2277 tonic::include_proto!("temporal.api.history.v1");
2278 pub use self::sdk_helpers::*;
2279 mod sdk_helpers {
2280 use super::*;
2281 use crate::protos::temporal::api::{
2282 enums::v1::EventType, history::v1::history_event::Attributes,
2283 };
2284 use anyhow::bail;
2285 use std::fmt::{Display, Formatter};
2286
2287 impl History {
2288 pub fn extract_run_id_from_start(&self) -> Result<&str, anyhow::Error> {
2289 extract_original_run_id_from_events(&self.events)
2290 }
2291
2292 pub fn last_event_id(&self) -> i64 {
2295 self.events.last().map(|e| e.event_id).unwrap_or_default()
2296 }
2297 }
2298
2299 pub fn extract_original_run_id_from_events(
2300 events: &[HistoryEvent],
2301 ) -> Result<&str, anyhow::Error> {
2302 if let Some(Attributes::WorkflowExecutionStartedEventAttributes(wes)) =
2303 events.get(0).and_then(|x| x.attributes.as_ref())
2304 {
2305 Ok(&wes.original_execution_run_id)
2306 } else {
2307 bail!("First event is not WorkflowExecutionStarted?!?")
2308 }
2309 }
2310
2311 impl HistoryEvent {
2312 pub fn is_command_event(&self) -> bool {
2314 EventType::try_from(self.event_type).map_or(false, |et| match et {
2315 EventType::ActivityTaskScheduled
2316 | EventType::ActivityTaskCancelRequested
2317 | EventType::MarkerRecorded
2318 | EventType::RequestCancelExternalWorkflowExecutionInitiated
2319 | EventType::SignalExternalWorkflowExecutionInitiated
2320 | EventType::StartChildWorkflowExecutionInitiated
2321 | EventType::TimerCanceled
2322 | EventType::TimerStarted
2323 | EventType::UpsertWorkflowSearchAttributes
2324 | EventType::WorkflowPropertiesModified
2325 | EventType::NexusOperationScheduled
2326 | EventType::NexusOperationCancelRequested
2327 | EventType::WorkflowExecutionCanceled
2328 | EventType::WorkflowExecutionCompleted
2329 | EventType::WorkflowExecutionContinuedAsNew
2330 | EventType::WorkflowExecutionFailed
2331 | EventType::WorkflowExecutionUpdateAccepted
2332 | EventType::WorkflowExecutionUpdateRejected
2333 | EventType::WorkflowExecutionUpdateCompleted => true,
2334 _ => false,
2335 })
2336 }
2337
2338 pub fn get_initial_command_event_id(&self) -> Option<i64> {
2342 self.attributes.as_ref().and_then(|a| {
2343 match a {
2346 Attributes::ActivityTaskStartedEventAttributes(a) =>
2347 Some(a.scheduled_event_id),
2348 Attributes::ActivityTaskCompletedEventAttributes(a) =>
2349 Some(a.scheduled_event_id),
2350 Attributes::ActivityTaskFailedEventAttributes(a) => Some(a.scheduled_event_id),
2351 Attributes::ActivityTaskTimedOutEventAttributes(a) => Some(a.scheduled_event_id),
2352 Attributes::ActivityTaskCancelRequestedEventAttributes(a) => Some(a.scheduled_event_id),
2353 Attributes::ActivityTaskCanceledEventAttributes(a) => Some(a.scheduled_event_id),
2354 Attributes::TimerFiredEventAttributes(a) => Some(a.started_event_id),
2355 Attributes::TimerCanceledEventAttributes(a) => Some(a.started_event_id),
2356 Attributes::RequestCancelExternalWorkflowExecutionFailedEventAttributes(a) => Some(a.initiated_event_id),
2357 Attributes::ExternalWorkflowExecutionCancelRequestedEventAttributes(a) => Some(a.initiated_event_id),
2358 Attributes::StartChildWorkflowExecutionFailedEventAttributes(a) => Some(a.initiated_event_id),
2359 Attributes::ChildWorkflowExecutionStartedEventAttributes(a) => Some(a.initiated_event_id),
2360 Attributes::ChildWorkflowExecutionCompletedEventAttributes(a) => Some(a.initiated_event_id),
2361 Attributes::ChildWorkflowExecutionFailedEventAttributes(a) => Some(a.initiated_event_id),
2362 Attributes::ChildWorkflowExecutionCanceledEventAttributes(a) => Some(a.initiated_event_id),
2363 Attributes::ChildWorkflowExecutionTimedOutEventAttributes(a) => Some(a.initiated_event_id),
2364 Attributes::ChildWorkflowExecutionTerminatedEventAttributes(a) => Some(a.initiated_event_id),
2365 Attributes::SignalExternalWorkflowExecutionFailedEventAttributes(a) => Some(a.initiated_event_id),
2366 Attributes::ExternalWorkflowExecutionSignaledEventAttributes(a) => Some(a.initiated_event_id),
2367 Attributes::WorkflowTaskStartedEventAttributes(a) => Some(a.scheduled_event_id),
2368 Attributes::WorkflowTaskCompletedEventAttributes(a) => Some(a.scheduled_event_id),
2369 Attributes::WorkflowTaskTimedOutEventAttributes(a) => Some(a.scheduled_event_id),
2370 Attributes::WorkflowTaskFailedEventAttributes(a) => Some(a.scheduled_event_id),
2371 Attributes::NexusOperationStartedEventAttributes(a) => Some(a.scheduled_event_id),
2372 Attributes::NexusOperationCompletedEventAttributes(a) => Some(a.scheduled_event_id),
2373 Attributes::NexusOperationFailedEventAttributes(a) => Some(a.scheduled_event_id),
2374 Attributes::NexusOperationTimedOutEventAttributes(a) => Some(a.scheduled_event_id),
2375 Attributes::NexusOperationCanceledEventAttributes(a) => Some(a.scheduled_event_id),
2376 Attributes::NexusOperationCancelRequestedEventAttributes(a) => Some(a.scheduled_event_id),
2377 Attributes::NexusOperationCancelRequestCompletedEventAttributes(a) => Some(a.scheduled_event_id),
2378 Attributes::NexusOperationCancelRequestFailedEventAttributes(a) => Some(a.scheduled_event_id),
2379 _ => None
2380 }
2381 })
2382 }
2383
2384 pub fn get_protocol_instance_id(&self) -> Option<&str> {
2386 self.attributes.as_ref().and_then(|a| match a {
2387 Attributes::WorkflowExecutionUpdateAcceptedEventAttributes(a) => {
2388 Some(a.protocol_instance_id.as_str())
2389 }
2390 _ => None,
2391 })
2392 }
2393
2394 pub fn is_final_wf_execution_event(&self) -> bool {
2396 match self.event_type() {
2397 EventType::WorkflowExecutionCompleted => true,
2398 EventType::WorkflowExecutionCanceled => true,
2399 EventType::WorkflowExecutionFailed => true,
2400 EventType::WorkflowExecutionTimedOut => true,
2401 EventType::WorkflowExecutionContinuedAsNew => true,
2402 EventType::WorkflowExecutionTerminated => true,
2403 _ => false,
2404 }
2405 }
2406
2407 pub fn is_wft_closed_event(&self) -> bool {
2408 match self.event_type() {
2409 EventType::WorkflowTaskCompleted => true,
2410 EventType::WorkflowTaskFailed => true,
2411 EventType::WorkflowTaskTimedOut => true,
2412 _ => false,
2413 }
2414 }
2415
2416 pub fn is_ignorable(&self) -> bool {
2417 if !self.worker_may_ignore {
2418 return false;
2419 }
2420 if let Some(a) = self.attributes.as_ref() {
2423 match a {
2424 Attributes::WorkflowExecutionStartedEventAttributes(_) => false,
2425 Attributes::WorkflowExecutionCompletedEventAttributes(_) => false,
2426 Attributes::WorkflowExecutionFailedEventAttributes(_) => false,
2427 Attributes::WorkflowExecutionTimedOutEventAttributes(_) => false,
2428 Attributes::WorkflowTaskScheduledEventAttributes(_) => false,
2429 Attributes::WorkflowTaskStartedEventAttributes(_) => false,
2430 Attributes::WorkflowTaskCompletedEventAttributes(_) => false,
2431 Attributes::WorkflowTaskTimedOutEventAttributes(_) => false,
2432 Attributes::WorkflowTaskFailedEventAttributes(_) => false,
2433 Attributes::ActivityTaskScheduledEventAttributes(_) => false,
2434 Attributes::ActivityTaskStartedEventAttributes(_) => false,
2435 Attributes::ActivityTaskCompletedEventAttributes(_) => false,
2436 Attributes::ActivityTaskFailedEventAttributes(_) => false,
2437 Attributes::ActivityTaskTimedOutEventAttributes(_) => false,
2438 Attributes::TimerStartedEventAttributes(_) => false,
2439 Attributes::TimerFiredEventAttributes(_) => false,
2440 Attributes::ActivityTaskCancelRequestedEventAttributes(_) => false,
2441 Attributes::ActivityTaskCanceledEventAttributes(_) => false,
2442 Attributes::TimerCanceledEventAttributes(_) => false,
2443 Attributes::MarkerRecordedEventAttributes(_) => false,
2444 Attributes::WorkflowExecutionSignaledEventAttributes(_) => false,
2445 Attributes::WorkflowExecutionTerminatedEventAttributes(_) => false,
2446 Attributes::WorkflowExecutionCancelRequestedEventAttributes(_) => false,
2447 Attributes::WorkflowExecutionCanceledEventAttributes(_) => false,
2448 Attributes::RequestCancelExternalWorkflowExecutionInitiatedEventAttributes(_) => false,
2449 Attributes::RequestCancelExternalWorkflowExecutionFailedEventAttributes(_) => false,
2450 Attributes::ExternalWorkflowExecutionCancelRequestedEventAttributes(_) => false,
2451 Attributes::WorkflowExecutionContinuedAsNewEventAttributes(_) => false,
2452 Attributes::StartChildWorkflowExecutionInitiatedEventAttributes(_) => false,
2453 Attributes::StartChildWorkflowExecutionFailedEventAttributes(_) => false,
2454 Attributes::ChildWorkflowExecutionStartedEventAttributes(_) => false,
2455 Attributes::ChildWorkflowExecutionCompletedEventAttributes(_) => false,
2456 Attributes::ChildWorkflowExecutionFailedEventAttributes(_) => false,
2457 Attributes::ChildWorkflowExecutionCanceledEventAttributes(_) => false,
2458 Attributes::ChildWorkflowExecutionTimedOutEventAttributes(_) => false,
2459 Attributes::ChildWorkflowExecutionTerminatedEventAttributes(_) => false,
2460 Attributes::SignalExternalWorkflowExecutionInitiatedEventAttributes(_) => false,
2461 Attributes::SignalExternalWorkflowExecutionFailedEventAttributes(_) => false,
2462 Attributes::ExternalWorkflowExecutionSignaledEventAttributes(_) => false,
2463 Attributes::UpsertWorkflowSearchAttributesEventAttributes(_) => false,
2464 Attributes::WorkflowExecutionUpdateAcceptedEventAttributes(_) => false,
2465 Attributes::WorkflowExecutionUpdateRejectedEventAttributes(_) => false,
2466 Attributes::WorkflowExecutionUpdateCompletedEventAttributes(_) => false,
2467 Attributes::WorkflowPropertiesModifiedExternallyEventAttributes(_) => false,
2468 Attributes::ActivityPropertiesModifiedExternallyEventAttributes(_) => false,
2469 Attributes::WorkflowPropertiesModifiedEventAttributes(_) => false,
2470 Attributes::WorkflowExecutionUpdateAdmittedEventAttributes(_) => false,
2471 Attributes::NexusOperationScheduledEventAttributes(_) => false,
2472 Attributes::NexusOperationStartedEventAttributes(_) => false,
2473 Attributes::NexusOperationCompletedEventAttributes(_) => false,
2474 Attributes::NexusOperationFailedEventAttributes(_) => false,
2475 Attributes::NexusOperationCanceledEventAttributes(_) => false,
2476 Attributes::NexusOperationTimedOutEventAttributes(_) => false,
2477 Attributes::NexusOperationCancelRequestedEventAttributes(_) => false,
2478 Attributes::WorkflowExecutionOptionsUpdatedEventAttributes(_) => true,
2480 Attributes::NexusOperationCancelRequestCompletedEventAttributes(_) => false,
2481 Attributes::NexusOperationCancelRequestFailedEventAttributes(_) => false,
2482 Attributes::WorkflowExecutionPausedEventAttributes(_) => true,
2484 Attributes::WorkflowExecutionUnpausedEventAttributes(_) => true,
2486 Attributes::WorkflowExecutionTimeSkippingTransitionedEventAttributes(_) => true,
2488 }
2489 } else {
2490 self.worker_may_ignore
2492 }
2493 }
2494 }
2495
2496 impl Display for HistoryEvent {
2497 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2498 write!(
2499 f,
2500 "HistoryEvent(id: {}, {:?})",
2501 self.event_id,
2502 EventType::try_from(self.event_type).unwrap_or_default()
2503 )
2504 }
2505 }
2506
2507 impl Attributes {
2508 pub fn event_type(&self) -> EventType {
2509 match self {
2511 Attributes::WorkflowExecutionStartedEventAttributes(_) => { EventType::WorkflowExecutionStarted }
2512 Attributes::WorkflowExecutionCompletedEventAttributes(_) => { EventType::WorkflowExecutionCompleted }
2513 Attributes::WorkflowExecutionFailedEventAttributes(_) => { EventType::WorkflowExecutionFailed }
2514 Attributes::WorkflowExecutionTimedOutEventAttributes(_) => { EventType::WorkflowExecutionTimedOut }
2515 Attributes::WorkflowTaskScheduledEventAttributes(_) => { EventType::WorkflowTaskScheduled }
2516 Attributes::WorkflowTaskStartedEventAttributes(_) => { EventType::WorkflowTaskStarted }
2517 Attributes::WorkflowTaskCompletedEventAttributes(_) => { EventType::WorkflowTaskCompleted }
2518 Attributes::WorkflowTaskTimedOutEventAttributes(_) => { EventType::WorkflowTaskTimedOut }
2519 Attributes::WorkflowTaskFailedEventAttributes(_) => { EventType::WorkflowTaskFailed }
2520 Attributes::ActivityTaskScheduledEventAttributes(_) => { EventType::ActivityTaskScheduled }
2521 Attributes::ActivityTaskStartedEventAttributes(_) => { EventType::ActivityTaskStarted }
2522 Attributes::ActivityTaskCompletedEventAttributes(_) => { EventType::ActivityTaskCompleted }
2523 Attributes::ActivityTaskFailedEventAttributes(_) => { EventType::ActivityTaskFailed }
2524 Attributes::ActivityTaskTimedOutEventAttributes(_) => { EventType::ActivityTaskTimedOut }
2525 Attributes::TimerStartedEventAttributes(_) => { EventType::TimerStarted }
2526 Attributes::TimerFiredEventAttributes(_) => { EventType::TimerFired }
2527 Attributes::ActivityTaskCancelRequestedEventAttributes(_) => { EventType::ActivityTaskCancelRequested }
2528 Attributes::ActivityTaskCanceledEventAttributes(_) => { EventType::ActivityTaskCanceled }
2529 Attributes::TimerCanceledEventAttributes(_) => { EventType::TimerCanceled }
2530 Attributes::MarkerRecordedEventAttributes(_) => { EventType::MarkerRecorded }
2531 Attributes::WorkflowExecutionSignaledEventAttributes(_) => { EventType::WorkflowExecutionSignaled }
2532 Attributes::WorkflowExecutionTerminatedEventAttributes(_) => { EventType::WorkflowExecutionTerminated }
2533 Attributes::WorkflowExecutionCancelRequestedEventAttributes(_) => { EventType::WorkflowExecutionCancelRequested }
2534 Attributes::WorkflowExecutionCanceledEventAttributes(_) => { EventType::WorkflowExecutionCanceled }
2535 Attributes::RequestCancelExternalWorkflowExecutionInitiatedEventAttributes(_) => { EventType::RequestCancelExternalWorkflowExecutionInitiated }
2536 Attributes::RequestCancelExternalWorkflowExecutionFailedEventAttributes(_) => { EventType::RequestCancelExternalWorkflowExecutionFailed }
2537 Attributes::ExternalWorkflowExecutionCancelRequestedEventAttributes(_) => { EventType::ExternalWorkflowExecutionCancelRequested }
2538 Attributes::WorkflowExecutionContinuedAsNewEventAttributes(_) => { EventType::WorkflowExecutionContinuedAsNew }
2539 Attributes::StartChildWorkflowExecutionInitiatedEventAttributes(_) => { EventType::StartChildWorkflowExecutionInitiated }
2540 Attributes::StartChildWorkflowExecutionFailedEventAttributes(_) => { EventType::StartChildWorkflowExecutionFailed }
2541 Attributes::ChildWorkflowExecutionStartedEventAttributes(_) => { EventType::ChildWorkflowExecutionStarted }
2542 Attributes::ChildWorkflowExecutionCompletedEventAttributes(_) => { EventType::ChildWorkflowExecutionCompleted }
2543 Attributes::ChildWorkflowExecutionFailedEventAttributes(_) => { EventType::ChildWorkflowExecutionFailed }
2544 Attributes::ChildWorkflowExecutionCanceledEventAttributes(_) => { EventType::ChildWorkflowExecutionCanceled }
2545 Attributes::ChildWorkflowExecutionTimedOutEventAttributes(_) => { EventType::ChildWorkflowExecutionTimedOut }
2546 Attributes::ChildWorkflowExecutionTerminatedEventAttributes(_) => { EventType::ChildWorkflowExecutionTerminated }
2547 Attributes::SignalExternalWorkflowExecutionInitiatedEventAttributes(_) => { EventType::SignalExternalWorkflowExecutionInitiated }
2548 Attributes::SignalExternalWorkflowExecutionFailedEventAttributes(_) => { EventType::SignalExternalWorkflowExecutionFailed }
2549 Attributes::ExternalWorkflowExecutionSignaledEventAttributes(_) => { EventType::ExternalWorkflowExecutionSignaled }
2550 Attributes::UpsertWorkflowSearchAttributesEventAttributes(_) => { EventType::UpsertWorkflowSearchAttributes }
2551 Attributes::WorkflowExecutionUpdateAdmittedEventAttributes(_) => { EventType::WorkflowExecutionUpdateAdmitted }
2552 Attributes::WorkflowExecutionUpdateRejectedEventAttributes(_) => { EventType::WorkflowExecutionUpdateRejected }
2553 Attributes::WorkflowExecutionUpdateAcceptedEventAttributes(_) => { EventType::WorkflowExecutionUpdateAccepted }
2554 Attributes::WorkflowExecutionUpdateCompletedEventAttributes(_) => { EventType::WorkflowExecutionUpdateCompleted }
2555 Attributes::WorkflowPropertiesModifiedExternallyEventAttributes(_) => { EventType::WorkflowPropertiesModifiedExternally }
2556 Attributes::ActivityPropertiesModifiedExternallyEventAttributes(_) => { EventType::ActivityPropertiesModifiedExternally }
2557 Attributes::WorkflowPropertiesModifiedEventAttributes(_) => { EventType::WorkflowPropertiesModified }
2558 Attributes::NexusOperationScheduledEventAttributes(_) => { EventType::NexusOperationScheduled }
2559 Attributes::NexusOperationStartedEventAttributes(_) => { EventType::NexusOperationStarted }
2560 Attributes::NexusOperationCompletedEventAttributes(_) => { EventType::NexusOperationCompleted }
2561 Attributes::NexusOperationFailedEventAttributes(_) => { EventType::NexusOperationFailed }
2562 Attributes::NexusOperationCanceledEventAttributes(_) => { EventType::NexusOperationCanceled }
2563 Attributes::NexusOperationTimedOutEventAttributes(_) => { EventType::NexusOperationTimedOut }
2564 Attributes::NexusOperationCancelRequestedEventAttributes(_) => { EventType::NexusOperationCancelRequested }
2565 Attributes::WorkflowExecutionOptionsUpdatedEventAttributes(_) => { EventType::WorkflowExecutionOptionsUpdated }
2566 Attributes::NexusOperationCancelRequestCompletedEventAttributes(_) => { EventType::NexusOperationCancelRequestCompleted }
2567 Attributes::NexusOperationCancelRequestFailedEventAttributes(_) => { EventType::NexusOperationCancelRequestFailed }
2568 Attributes::WorkflowExecutionPausedEventAttributes(_) => { EventType::WorkflowExecutionPaused }
2569 Attributes::WorkflowExecutionUnpausedEventAttributes(_) => { EventType::WorkflowExecutionUnpaused }
2570 Attributes::WorkflowExecutionTimeSkippingTransitionedEventAttributes(_) => { EventType::WorkflowExecutionTimeSkippingTransitioned }
2571 }
2572 }
2573 }
2574 }
2575 }
2576 }
2577 pub mod namespace {
2578 pub mod v1 {
2579 tonic::include_proto!("temporal.api.namespace.v1");
2580 }
2581 }
2582 pub mod operatorservice {
2583 pub mod v1 {
2584 tonic::include_proto!("temporal.api.operatorservice.v1");
2585 }
2586 }
2587 pub mod protocol {
2588 pub mod v1 {
2589 tonic::include_proto!("temporal.api.protocol.v1");
2590 mod sdk_helpers {
2591 use super::*;
2592 use std::fmt::{Display, Formatter};
2593
2594 impl Display for Message {
2595 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2596 write!(f, "ProtocolMessage({})", self.id)
2597 }
2598 }
2599 }
2600 }
2601 }
2602 pub mod query {
2603 pub mod v1 {
2604 tonic::include_proto!("temporal.api.query.v1");
2605 }
2606 }
2607 pub mod replication {
2608 pub mod v1 {
2609 tonic::include_proto!("temporal.api.replication.v1");
2610 }
2611 }
2612 pub mod rules {
2613 pub mod v1 {
2614 tonic::include_proto!("temporal.api.rules.v1");
2615 }
2616 }
2617 pub mod schedule {
2618 #[allow(rustdoc::invalid_html_tags)]
2619 pub mod v1 {
2620 tonic::include_proto!("temporal.api.schedule.v1");
2621 }
2622 }
2623 pub mod sdk {
2624 pub mod v1 {
2625 tonic::include_proto!("temporal.api.sdk.v1");
2626 }
2627 }
2628 pub mod taskqueue {
2629 pub mod v1 {
2630 tonic::include_proto!("temporal.api.taskqueue.v1");
2631 mod sdk_helpers {
2632 use super::*;
2633 use crate::protos::temporal::api::enums::v1::TaskQueueKind;
2634
2635 impl From<String> for TaskQueue {
2636 fn from(name: String) -> Self {
2637 Self {
2638 name,
2639 kind: TaskQueueKind::Normal as i32,
2640 normal_name: "".to_string(),
2641 }
2642 }
2643 }
2644 }
2645 }
2646 }
2647 pub mod testservice {
2648 pub mod v1 {
2649 tonic::include_proto!("temporal.api.testservice.v1");
2650 }
2651 }
2652 pub mod update {
2653 pub mod v1 {
2654 tonic::include_proto!("temporal.api.update.v1");
2655 mod sdk_helpers {
2656 use super::*;
2657 use crate::protos::temporal::api::update::v1::outcome::Value;
2658
2659 impl Outcome {
2660 pub fn is_success(&self) -> bool {
2661 match self.value {
2662 Some(Value::Success(_)) => true,
2663 _ => false,
2664 }
2665 }
2666 }
2667 }
2668 }
2669 }
2670 pub mod version {
2671 pub mod v1 {
2672 tonic::include_proto!("temporal.api.version.v1");
2673 }
2674 }
2675 pub mod worker {
2676 pub mod v1 {
2677 tonic::include_proto!("temporal.api.worker.v1");
2678 }
2679 }
2680 pub mod workflow {
2681 pub mod v1 {
2682 tonic::include_proto!("temporal.api.workflow.v1");
2683 }
2684 }
2685 pub mod nexus {
2686 pub mod v1 {
2687 tonic::include_proto!("temporal.api.nexus.v1");
2688 pub use self::sdk_helpers::*;
2689 mod sdk_helpers {
2690 use super::*;
2691 use crate::protos::{
2692 camel_case_to_screaming_snake,
2693 temporal::api::{
2694 common::{
2695 self,
2696 v1::link::{WorkflowEvent, workflow_event},
2697 },
2698 enums::v1::EventType,
2699 failure,
2700 },
2701 };
2702 use anyhow::{anyhow, bail};
2703 use http::Uri;
2704 #[cfg(feature = "serde_serialize")]
2705 use prost::Name;
2706 #[cfg(feature = "serde_serialize")]
2707 use std::collections::HashMap;
2708 use std::fmt::{Display, Formatter};
2709
2710 impl Display for Response {
2711 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2712 write!(f, "NexusResponse(",)?;
2713 match &self.variant {
2714 None => {}
2715 Some(v) => {
2716 write!(f, "{v}")?;
2717 }
2718 }
2719 write!(f, ")")
2720 }
2721 }
2722
2723 impl Display for response::Variant {
2724 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2725 match self {
2726 response::Variant::StartOperation(_) => {
2727 write!(f, "StartOperation")
2728 }
2729 response::Variant::CancelOperation(_) => {
2730 write!(f, "CancelOperation")
2731 }
2732 }
2733 }
2734 }
2735
2736 impl Display for HandlerError {
2737 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2738 write!(f, "HandlerError")
2739 }
2740 }
2741
2742 pub enum NexusTaskFailure {
2743 Legacy(HandlerError),
2744 Temporal(failure::v1::Failure),
2745 }
2746
2747 static SCHEME_PREFIX: &str = "temporal://";
2748
2749 pub fn workflow_event_link_from_nexus(
2751 l: &Link,
2752 ) -> Result<common::v1::Link, anyhow::Error> {
2753 if !l.url.starts_with(SCHEME_PREFIX) {
2754 bail!("Invalid scheme for nexus link: {:?}", l.url);
2755 }
2756 let no_authority_url = l.url.strip_prefix(SCHEME_PREFIX).unwrap();
2759 let uri = Uri::try_from(no_authority_url)?;
2760 let parts = uri.into_parts();
2761 let path = parts.path_and_query.ok_or_else(|| {
2762 anyhow!("Failed to parse nexus link, invalid path: {:?}", l)
2763 })?;
2764 let path_parts = path.path().split('/').collect::<Vec<_>>();
2765 if path_parts.get(1) != Some(&"namespaces") {
2766 bail!("Invalid path for nexus link: {:?}", l);
2767 }
2768 let namespace = path_parts.get(2).ok_or_else(|| {
2769 anyhow!("Failed to parse nexus link, no namespace: {:?}", l)
2770 })?;
2771 if path_parts.get(3) != Some(&"workflows") {
2772 bail!("Invalid path for nexus link, no workflows segment: {:?}", l);
2773 }
2774 let workflow_id = path_parts.get(4).ok_or_else(|| {
2775 anyhow!("Failed to parse nexus link, no workflow id: {:?}", l)
2776 })?;
2777 let run_id = path_parts.get(5).ok_or_else(|| {
2778 anyhow!("Failed to parse nexus link, no run id: {:?}", l)
2779 })?;
2780 if path_parts.get(6) != Some(&"history") {
2781 bail!("Invalid path for nexus link, no history segment: {:?}", l);
2782 }
2783 let reference = if let Some(query) = path.query() {
2784 let mut eventref = workflow_event::EventReference::default();
2785 let query_parts = query.split('&').collect::<Vec<_>>();
2786 for qp in query_parts {
2787 let mut kv = qp.split('=');
2788 let key = kv.next().ok_or_else(|| {
2789 anyhow!("Failed to parse nexus link query parameter: {:?}", l)
2790 })?;
2791 let val = kv.next().ok_or_else(|| {
2792 anyhow!("Failed to parse nexus link query parameter: {:?}", l)
2793 })?;
2794 match key {
2795 "eventID" => {
2796 eventref.event_id = val.parse().map_err(|_| {
2797 anyhow!("Failed to parse nexus link event id: {:?}", l)
2798 })?;
2799 }
2800 "eventType" => {
2801 eventref.event_type = EventType::from_str_name(val)
2802 .unwrap_or_else(|| {
2803 EventType::from_str_name(
2804 &("EVENT_TYPE_".to_string()
2805 + &camel_case_to_screaming_snake(val)),
2806 )
2807 .unwrap_or_default()
2808 })
2809 .into()
2810 }
2811 _ => continue,
2812 }
2813 }
2814 Some(workflow_event::Reference::EventRef(eventref))
2815 } else {
2816 None
2817 };
2818
2819 Ok(common::v1::Link {
2820 variant: Some(common::v1::link::Variant::WorkflowEvent(
2821 WorkflowEvent {
2822 namespace: namespace.to_string(),
2823 workflow_id: workflow_id.to_string(),
2824 run_id: run_id.to_string(),
2825 reference,
2826 },
2827 )),
2828 })
2829 }
2830
2831 #[cfg(feature = "serde_serialize")]
2832 impl TryFrom<failure::v1::Failure> for Failure {
2833 type Error = serde_json::Error;
2834
2835 fn try_from(mut f: failure::v1::Failure) -> Result<Self, Self::Error> {
2836 let message = std::mem::take(&mut f.message);
2838
2839 let details = serde_json::to_vec(&f)?;
2841
2842 Ok(Failure {
2844 message,
2845 stack_trace: f.stack_trace,
2846 metadata: HashMap::from([(
2847 "type".to_string(),
2848 failure::v1::Failure::full_name().into(),
2849 )]),
2850 details,
2851 cause: None,
2852 })
2853 }
2854 }
2855 }
2856 }
2857 }
2858 pub mod nexusservices {
2859 pub mod workerservice {
2860 pub mod v1 {
2861 tonic::include_proto!("temporal.api.nexusservices.workerservice.v1");
2862 }
2863 }
2864 }
2865 pub mod workflowservice {
2866 pub mod v1 {
2867 tonic::include_proto!("temporal.api.workflowservice.v1");
2868 pub use self::sdk_helpers::*;
2869 mod sdk_helpers {
2870 use super::*;
2871 use std::{
2872 convert::TryInto,
2873 fmt::{Display, Formatter},
2874 time::{Duration, SystemTime},
2875 };
2876
2877 macro_rules! sched_to_start_impl {
2878 ($sched_field:ident) => {
2879 pub fn sched_to_start(&self) -> Option<Duration> {
2882 if let Some((sch, st)) =
2883 self.$sched_field.clone().zip(self.started_time.clone())
2884 {
2885 if let Some(value) = elapsed_between_prost_times(sch, st) {
2886 return value;
2887 }
2888 }
2889 None
2890 }
2891 };
2892 }
2893
2894 fn elapsed_between_prost_times(
2895 from: prost_types::Timestamp,
2896 to: prost_types::Timestamp,
2897 ) -> Option<Option<Duration>> {
2898 let from: Result<SystemTime, _> = from.try_into();
2899 let to: Result<SystemTime, _> = to.try_into();
2900 if let (Ok(from), Ok(to)) = (from, to) {
2901 return Some(to.duration_since(from).ok());
2902 }
2903 None
2904 }
2905
2906 impl PollWorkflowTaskQueueResponse {
2907 sched_to_start_impl!(scheduled_time);
2908 }
2909
2910 impl Display for PollWorkflowTaskQueueResponse {
2911 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2912 let last_event = self
2913 .history
2914 .as_ref()
2915 .and_then(|h| h.events.last().map(|he| he.event_id))
2916 .unwrap_or(0);
2917 write!(
2918 f,
2919 "PollWFTQResp(run_id: {}, attempt: {}, last_event: {})",
2920 self.workflow_execution
2921 .as_ref()
2922 .map_or("", |we| we.run_id.as_str()),
2923 self.attempt,
2924 last_event
2925 )
2926 }
2927 }
2928
2929 pub struct CompactHist<'a>(pub &'a PollWorkflowTaskQueueResponse);
2931 impl Display for CompactHist<'_> {
2932 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
2933 writeln!(
2934 f,
2935 "PollWorkflowTaskQueueResponse (prev_started: {}, started: {})",
2936 self.0.previous_started_event_id, self.0.started_event_id
2937 )?;
2938 if let Some(h) = self.0.history.as_ref() {
2939 for event in &h.events {
2940 writeln!(f, "{}", event)?;
2941 }
2942 }
2943 writeln!(f, "query: {:#?}", self.0.query)?;
2944 writeln!(f, "queries: {:#?}", self.0.queries)
2945 }
2946 }
2947
2948 impl PollActivityTaskQueueResponse {
2949 sched_to_start_impl!(current_attempt_scheduled_time);
2950 }
2951
2952 impl PollNexusTaskQueueResponse {
2953 pub fn sched_to_start(&self) -> Option<Duration> {
2954 if let Some((sch, st)) = self
2955 .request
2956 .as_ref()
2957 .and_then(|r| r.scheduled_time)
2958 .clone()
2959 .zip(SystemTime::now().try_into().ok())
2960 {
2961 if let Some(value) = elapsed_between_prost_times(sch, st) {
2962 return value;
2963 }
2964 }
2965 None
2966 }
2967 }
2968
2969 impl QueryWorkflowResponse {
2970 pub fn unwrap(
2972 self,
2973 ) -> Vec<crate::protos::temporal::api::common::v1::Payload>
2974 {
2975 self.query_result.unwrap().payloads
2976 }
2977 }
2978 }
2979 }
2980 }
2981 }
2982}
2983
2984#[allow(
2985 clippy::all,
2986 missing_docs,
2987 rustdoc::broken_intra_doc_links,
2988 rustdoc::bare_urls
2989)]
2990pub mod google {
2991 pub mod rpc {
2992 tonic::include_proto!("google.rpc");
2993 }
2994}
2995
2996#[allow(
2997 clippy::all,
2998 missing_docs,
2999 rustdoc::broken_intra_doc_links,
3000 rustdoc::bare_urls
3001)]
3002pub mod grpc {
3003 pub mod health {
3004 pub mod v1 {
3005 tonic::include_proto!("grpc.health.v1");
3006 }
3007 }
3008}
3009mod sdk_helpers {
3010 use std::time::Duration;
3011
3012 pub fn camel_case_to_screaming_snake(val: &str) -> String {
3014 let mut out = String::new();
3015 let mut last_was_upper = true;
3016 for c in val.chars() {
3017 if c.is_uppercase() {
3018 if !last_was_upper {
3019 out.push('_');
3020 }
3021 out.push(c.to_ascii_uppercase());
3022 last_was_upper = true;
3023 } else {
3024 out.push(c.to_ascii_uppercase());
3025 last_was_upper = false;
3026 }
3027 }
3028 out
3029 }
3030
3031 pub fn proto_ts_to_system_time(ts: &prost_types::Timestamp) -> Option<std::time::SystemTime> {
3033 std::time::SystemTime::UNIX_EPOCH.checked_add(
3034 Duration::from_secs(ts.seconds as u64) + Duration::from_nanos(ts.nanos as u64),
3035 )
3036 }
3037
3038 #[cfg(test)]
3039 mod tests {
3040 use crate::protos::{
3041 coresdk::{activity_task, activity_task::ActivityTask},
3042 temporal::api::{
3043 failure::v1::Failure, workflowservice::v1::PollActivityTaskQueueResponse,
3044 },
3045 };
3046 use anyhow::anyhow;
3047
3048 #[test]
3049 fn start_from_poll_resp_standalone_activity_populates_run_id() {
3050 let resp = PollActivityTaskQueueResponse {
3051 task_token: vec![1, 2, 3],
3052 activity_run_id: "test-run-id-123".to_string(),
3053 activity_id: "my-activity".to_string(),
3054 ..Default::default()
3055 };
3056 let task = ActivityTask::start_from_poll_resp(resp);
3057 let start = match task.variant {
3058 Some(activity_task::activity_task::Variant::Start(s)) => s,
3059 _ => panic!("expected Start variant"),
3060 };
3061 assert_eq!(start.run_id, "test-run-id-123");
3062 assert!(!start.is_local);
3063 }
3064
3065 #[test]
3066 fn start_from_poll_resp_workflow_activity_has_empty_run_id() {
3067 use crate::protos::temporal::api::common::v1::WorkflowExecution;
3068 let resp = PollActivityTaskQueueResponse {
3069 task_token: vec![4, 5, 6],
3070 activity_id: "my-workflow-activity".to_string(),
3071 workflow_execution: Some(WorkflowExecution {
3072 workflow_id: "wf-123".to_string(),
3073 run_id: "wf-run-456".to_string(),
3074 }),
3075 ..Default::default()
3077 };
3078 let task = ActivityTask::start_from_poll_resp(resp);
3079 let start = match task.variant {
3080 Some(activity_task::activity_task::Variant::Start(s)) => s,
3081 _ => panic!("expected Start variant"),
3082 };
3083 assert!(start.run_id.is_empty());
3084 assert_eq!(start.workflow_execution.unwrap().run_id, "wf-run-456");
3086 }
3087
3088 #[test]
3089 fn anyhow_to_failure_conversion() {
3090 let no_causes: Failure = anyhow!("no causes").into();
3091 assert_eq!(no_causes.cause, None);
3092 assert_eq!(no_causes.message, "no causes");
3093 let orig = anyhow!("fail 1");
3094 let mid = orig.context("fail 2");
3095 let top = mid.context("fail 3");
3096 let as_fail: Failure = top.into();
3097 assert_eq!(as_fail.message, "fail 3");
3098 assert_eq!(as_fail.cause.as_ref().unwrap().message, "fail 2");
3099 assert_eq!(as_fail.cause.unwrap().cause.unwrap().message, "fail 1");
3100 }
3101 }
3102}
3103pub use self::sdk_helpers::*;