Updated Activity/Workflow Function bounds to allow a more generic user function

Author: h7kanna

core/src/core_tests/local_activities.rs

@@ -21,8 +21,8 @@ use std::{
 };
 use temporal_client::WorkflowOptions;
 use temporal_sdk::{
-    ActContext, ActivityCancelledError, LocalActivityOptions, WfContext, WorkflowFunction,
-    WorkflowResult,
+    ActContext, ActExitValue, ActivityCancelledError, ActivityFunction, LocalActivityOptions,
+    WfContext, WorkflowFunction, WorkflowResult,
 };
 use temporal_sdk_core_api::{
     errors::{PollActivityError, PollWfError},
@@ -101,7 +101,7 @@ async fn local_act_two_wfts_before_marker(#[case] replay: bool, #[case] cached:
             Ok(().into())
         },
     );
-    worker.register_activity(DEFAULT_ACTIVITY_TYPE, echo);
+    worker.register_activity(DEFAULT_ACTIVITY_TYPE, ActivityFunction::from(echo));
     worker
         .submit_wf(
             wf_id.to_owned(),
@@ -151,7 +151,7 @@ async fn local_act_many_concurrent() {
     let mut worker = mock_sdk(mh);
 
     worker.register_wf(DEFAULT_WORKFLOW_TYPE.to_owned(), local_act_fanout_wf);
-    worker.register_activity("echo", echo);
+    worker.register_activity("echo", ActivityFunction::from(echo));
     worker
         .submit_wf(
             wf_id.to_owned(),
@@ -207,14 +207,17 @@ async fn local_act_heartbeat(#[case] shutdown_middle: bool) {
             Ok(().into())
         },
     );
-    worker.register_activity("echo", move |_ctx: ActContext, str: String| async move {
-        if shutdown_middle {
-            shutdown_barr.wait().await;
-        }
-        // Take slightly more than two workflow tasks
-        tokio::time::sleep(wft_timeout.mul_f32(2.2)).await;
-        Ok(str)
-    });
+    worker.register_activity(
+        "echo",
+        ActivityFunction::from(move |_ctx: ActContext, str: String| async move {
+            if shutdown_middle {
+                shutdown_barr.wait().await;
+            }
+            // Take slightly more than two workflow tasks
+            tokio::time::sleep(wft_timeout.mul_f32(2.2)).await;
+            Ok(str)
+        }),
+    );
     worker
         .submit_wf(
             wf_id.to_owned(),
@@ -276,10 +279,10 @@ async fn local_act_fail_and_retry(#[case] eventually_pass: bool) {
         },
     );
     let attempts: &'static _ = Box::leak(Box::new(AtomicUsize::new(0)));
-    worker.register_activity("echo", move |_ctx: ActContext, _: String| async move {
+    worker.register_activity("echo", move |_ctx: ActContext| async move {
         // Succeed on 3rd attempt (which is ==2 since fetch_add returns prev val)
         if 2 == attempts.fetch_add(1, Ordering::Relaxed) && eventually_pass {
-            Ok(())
+            Ok(().into())
         } else {
             Err(anyhow!("Oh no I failed!"))
         }
@@ -355,12 +358,9 @@ async fn local_act_retry_long_backoff_uses_timer() {
             Ok(().into())
         },
     );
-    worker.register_activity(
-        DEFAULT_ACTIVITY_TYPE,
-        move |_ctx: ActContext, _: String| async move {
-            Result::<(), _>::Err(anyhow!("Oh no I failed!"))
-        },
-    );
+    worker.register_activity(DEFAULT_ACTIVITY_TYPE, move |_ctx: ActContext| async move {
+        Result::<ActExitValue<()>, _>::Err(anyhow!("Oh no I failed!"))
+    });
     worker
         .submit_wf(
             wf_id.to_owned(),
@@ -398,7 +398,7 @@ async fn local_act_null_result() {
             Ok(().into())
         },
     );
-    worker.register_activity("nullres", |_ctx: ActContext, _: String| async { Ok(()) });
+    worker.register_activity("nullres", |_ctx: ActContext| async { Ok(().into()) });
     worker
         .submit_wf(
             wf_id.to_owned(),
@@ -442,7 +442,7 @@ async fn local_act_command_immediately_follows_la_marker() {
             Ok(().into())
         },
     );
-    worker.register_activity("nullres", |_ctx: ActContext, _: String| async { Ok(()) });
+    worker.register_activity("nullres", |_ctx: ActContext| async { Ok(().into()) });
     worker
         .submit_wf(
             wf_id.to_owned(),
@@ -778,10 +778,7 @@ async fn test_schedule_to_start_timeout() {
             Ok(().into())
         },
     );
-    worker.register_activity(
-        "echo",
-        move |_ctx: ActContext, _: String| async move { Ok(()) },
-    );
+    worker.register_activity("echo", move |_ctx: ActContext| async move { Ok(().into()) });
     worker
         .submit_wf(
             wf_id.to_owned(),
@@ -868,10 +865,7 @@ async fn test_schedule_to_start_timeout_not_based_on_original_time(
             Ok(().into())
         },
     );
-    worker.register_activity(
-        "echo",
-        move |_ctx: ActContext, _: String| async move { Ok(()) },
-    );
+    worker.register_activity("echo", move |_ctx: ActContext| async move { Ok(().into()) });
     worker
         .submit_wf(
             wf_id.to_owned(),
@@ -917,16 +911,13 @@ async fn wft_failure_cancels_running_las() {
             Ok(().into())
         },
     );
-    worker.register_activity(
-        DEFAULT_ACTIVITY_TYPE,
-        move |ctx: ActContext, _: String| async move {
-            let res = tokio::time::timeout(Duration::from_millis(500), ctx.cancelled()).await;
-            if res.is_err() {
-                panic!("Activity must be cancelled!!!!");
-            }
-            Result::<(), _>::Err(ActivityCancelledError::default().into())
-        },
-    );
+    worker.register_activity(DEFAULT_ACTIVITY_TYPE, move |ctx: ActContext| async move {
+        let res = tokio::time::timeout(Duration::from_millis(500), ctx.cancelled()).await;
+        if res.is_err() {
+            panic!("Activity must be cancelled!!!!");
+        }
+        Result::<ActExitValue<()>, _>::Err(ActivityCancelledError::default().into())
+    });
     worker
         .submit_wf(
             wf_id.to_owned(),
@@ -980,7 +971,7 @@ async fn resolved_las_not_recorded_if_wft_fails_many_times() {
     );
     worker.register_activity(
         "echo",
-        move |_: ActContext, _: String| async move { Ok(()) },
+        ActivityFunction::from(move |_: ActContext, _: String| async move { Ok(()) }),
     );
     worker
         .submit_wf(
@@ -1039,9 +1030,12 @@ async fn local_act_records_nonfirst_attempts_ok() {
             Ok(().into())
         },
     );
-    worker.register_activity("echo", move |_ctx: ActContext, _: String| async move {
-        Result::<(), _>::Err(anyhow!("I fail"))
-    });
+    worker.register_activity(
+        "echo",
+        ActivityFunction::from(move |_ctx: ActContext, _: String| async move {
+            Result::<(), _>::Err(anyhow!("I fail"))
+        }),
+    );
     worker
         .submit_wf(
             wf_id.to_owned(),

sdk/src/activity_context.rs

@@ -60,15 +60,15 @@ impl ActContext {
         task_queue: String,
         task_token: Vec<u8>,
         task: activity_task::Start,
-    ) -> (Self, Payload) {
+    ) -> Self {
         let activity_task::Start {
             workflow_namespace,
             workflow_type,
             workflow_execution,
             activity_id,
             activity_type,
             header_fields,
-            mut input,
+            input,
             heartbeat_details,
             scheduled_time,
             current_attempt_scheduled_time,
@@ -86,37 +86,32 @@ impl ActContext {
             start_to_close_timeout.as_ref(),
             schedule_to_close_timeout.as_ref(),
         );
-        let first_arg = input.pop().unwrap_or_default();
 
-        (
-            ActContext {
-                worker,
-                app_data,
-                cancellation_token,
-                input,
-                heartbeat_details,
-                header_fields,
-                info: ActivityInfo {
-                    task_token,
-                    task_queue,
-                    workflow_type,
-                    workflow_namespace,
-                    workflow_execution,
-                    activity_id,
-                    activity_type,
-                    heartbeat_timeout: heartbeat_timeout.try_into_or_none(),
-                    scheduled_time: scheduled_time.try_into_or_none(),
-                    started_time: started_time.try_into_or_none(),
-                    deadline,
-                    attempt,
-                    current_attempt_scheduled_time: current_attempt_scheduled_time
-                        .try_into_or_none(),
-                    retry_policy,
-                    is_local,
-                },
+        ActContext {
+            worker,
+            app_data,
+            cancellation_token,
+            input,
+            heartbeat_details,
+            header_fields,
+            info: ActivityInfo {
+                task_token,
+                task_queue,
+                workflow_type,
+                workflow_namespace,
+                workflow_execution,
+                activity_id,
+                activity_type,
+                heartbeat_timeout: heartbeat_timeout.try_into_or_none(),
+                scheduled_time: scheduled_time.try_into_or_none(),
+                started_time: started_time.try_into_or_none(),
+                deadline,
+                attempt,
+                current_attempt_scheduled_time: current_attempt_scheduled_time.try_into_or_none(),
+                retry_policy,
+                is_local,
             },
-            first_arg,
-        )
+        }
     }
 
     /// Returns a future the completes if and when the activity this was called inside has been
@@ -133,8 +128,8 @@ impl ActContext {
     /// Retrieve extra parameters to the Activity. The first input is always popped and passed to
     /// the Activity function for the currently executing activity. However, if more parameters are
     /// passed, perhaps from another language's SDK, explicit access is available from extra_inputs
-    pub fn extra_inputs(&mut self) -> &mut [Payload] {
-        &mut self.input
+    pub fn get_args(&self) -> &[Payload] {
+        &self.input
     }
 
     /// Extract heartbeat details from last failed attempt. This is used in combination with retry policy.