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