Rename TestRustWorker to Worker

Author: Sushisource

client/Cargo.toml

@@ -36,4 +36,3 @@ uuid = { version = "0.8.2", features = ["v4"] }
 [dependencies.temporal-sdk-core-protos]
 path = "../sdk-core-protos"
 version = "0.1"
-

core/benches/workflow_replay.rs

@@ -1,7 +1,7 @@
 use criterion::{criterion_group, criterion_main, Criterion};
 use futures::StreamExt;
 use std::time::Duration;
-use temporal_sdk::{TestRustWorker, WfContext, WorkflowFunction};
+use temporal_sdk::{WfContext, Worker, WorkflowFunction};
 use temporal_sdk_core::{telemetry_init, TelemetryOptionsBuilder};
 use temporal_sdk_core_protos::DEFAULT_WORKFLOW_TYPE;
 use temporal_sdk_core_test_utils::{canned_histories, init_core_replay_preloaded};
@@ -29,7 +29,7 @@ pub fn criterion_benchmark(c: &mut Criterion) {
             tokio_runtime.block_on(async {
                 let func = timers_wf(num_timers);
                 let (worker, _) = init_core_replay_preloaded("replay_bench", &hist);
-                let mut worker = TestRustWorker::new(worker, "replay_bench".to_string(), None);
+                let mut worker = Worker::new(worker, "replay_bench".to_string(), None);
                 worker.register_wf(DEFAULT_WORKFLOW_TYPE, func);
                 worker.incr_expected_run_count(1);
                 worker.run_until_done().await.unwrap();
@@ -46,7 +46,7 @@ pub fn criterion_benchmark(c: &mut Criterion) {
             tokio_runtime.block_on(async {
                 let func = big_signals_wf(num_tasks);
                 let (worker, _) = init_core_replay_preloaded("large_hist_bench", &hist);
-                let mut worker = TestRustWorker::new(worker, "large_hist_bench".to_string(), None);
+                let mut worker = Worker::new(worker, "large_hist_bench".to_string(), None);
                 worker.register_wf(DEFAULT_WORKFLOW_TYPE, func);
                 worker.incr_expected_run_count(1);
                 worker.run_until_done().await.unwrap();

core/src/core_tests/child_workflows.rs

@@ -8,7 +8,7 @@ use crate::{
 use std::sync::Arc;
 use temporal_client::{mocks::mock_gateway, WorkflowOptions};
 use temporal_sdk::{
-    ChildWorkflowOptions, Signal, TestRustWorker, WfContext, WorkflowFunction, WorkflowResult,
+    ChildWorkflowOptions, Signal, WfContext, Worker, WorkflowFunction, WorkflowResult,
 };
 use temporal_sdk_core_protos::coresdk::child_workflow::{
     child_workflow_result, ChildWorkflowCancellationType,
@@ -29,7 +29,7 @@ async fn signal_child_workflow(#[case] serial: bool) {
     let mh = MockPollCfg::from_resp_batches(wf_id, t, [ResponseType::AllHistory], mock);
     let mock = build_mock_pollers(mh);
     let core = mock_worker(mock);
-    let mut worker = TestRustWorker::new(Arc::new(core), TEST_Q.to_string(), None);
+    let mut worker = Worker::new(Arc::new(core), TEST_Q.to_string(), None);
 
     let wf = move |ctx: WfContext| async move {
         let child = ctx.child_workflow(ChildWorkflowOptions {

core/src/core_tests/determinism.rs

@@ -12,7 +12,7 @@ use std::{
     time::Duration,
 };
 use temporal_client::{mocks::mock_gateway, WorkflowOptions};
-use temporal_sdk::{TestRustWorker, WfContext, WorkflowResult};
+use temporal_sdk::{WfContext, Worker, WorkflowResult};
 use temporal_sdk_core_protos::temporal::api::enums::v1::WorkflowTaskFailedCause;
 
 static DID_FAIL: AtomicBool = AtomicBool::new(false);
@@ -43,7 +43,7 @@ async fn test_wf_task_rejected_properly() {
         Box::new(|_, cause, _| matches!(cause, WorkflowTaskFailedCause::Unspecified));
     let mock = build_mock_pollers(mh);
     let core = mock_worker(mock);
-    let mut worker = TestRustWorker::new(Arc::new(core), TEST_Q.to_string(), None);
+    let mut worker = Worker::new(Arc::new(core), TEST_Q.to_string(), None);
 
     worker.register_wf(wf_type.to_owned(), timer_wf_fails_once);
     worker
@@ -87,7 +87,7 @@ async fn test_wf_task_rejected_properly_due_to_nondeterminism(#[case] use_cache:
         });
     }
     let core = mock_worker(mock);
-    let mut worker = TestRustWorker::new(Arc::new(core), TEST_Q.to_string(), None);
+    let mut worker = Worker::new(Arc::new(core), TEST_Q.to_string(), None);
 
     let started_count: &'static _ = Box::leak(Box::new(AtomicUsize::new(0)));
     worker.register_wf(wf_type.to_owned(), move |ctx: WfContext| async move {

core/src/core_tests/local_activities.rs

@@ -12,7 +12,7 @@ use std::{
     time::Duration,
 };
 use temporal_client::{mocks::mock_gateway, WorkflowOptions};
-use temporal_sdk::{LocalActivityOptions, TestRustWorker, WfContext, WorkflowResult};
+use temporal_sdk::{LocalActivityOptions, WfContext, Worker, WorkflowResult};
 use temporal_sdk_core_protos::{
     coresdk::{common::RetryPolicy, AsJsonPayloadExt},
     temporal::api::{enums::v1::EventType, failure::v1::Failure},
@@ -55,7 +55,7 @@ async fn local_act_two_wfts_before_marker(#[case] replay: bool, #[case] cached:
         mock.worker_cfg(|cfg| cfg.max_cached_workflows = 1);
     }
     let core = mock_worker(mock);
-    let mut worker = TestRustWorker::new(Arc::new(core), TEST_Q.to_string(), None);
+    let mut worker = Worker::new(Arc::new(core), TEST_Q.to_string(), None);
 
     worker.register_wf(
         DEFAULT_WORKFLOW_TYPE.to_owned(),
@@ -119,7 +119,7 @@ async fn local_act_many_concurrent() {
     let mh = MockPollCfg::from_resp_batches(wf_id, t, [1, 2, 3], mock);
     let mock = build_mock_pollers(mh);
     let core = mock_worker(mock);
-    let mut worker = TestRustWorker::new(Arc::new(core), TEST_Q.to_string(), None);
+    let mut worker = Worker::new(Arc::new(core), TEST_Q.to_string(), None);
 
     worker.register_wf(DEFAULT_WORKFLOW_TYPE.to_owned(), local_act_fanout_wf);
     worker.register_activity("echo", |str: String| async move { Ok(str) });
@@ -168,7 +168,7 @@ async fn local_act_heartbeat(#[case] shutdown_middle: bool) {
     let mut mock = build_mock_pollers(mh);
     mock.worker_cfg(|wc| wc.max_cached_workflows = 1);
     let core = Arc::new(mock_worker(mock));
-    let mut worker = TestRustWorker::new(core.clone(), TEST_Q.to_string(), None);
+    let mut worker = Worker::new(core.clone(), TEST_Q.to_string(), None);
     let shutdown_barr: &'static Barrier = Box::leak(Box::new(Barrier::new(2)));
 
     worker.register_wf(
@@ -226,7 +226,7 @@ async fn local_act_fail_and_retry(#[case] eventually_pass: bool) {
     let mh = MockPollCfg::from_resp_batches(wf_id, t, [1], mock);
     let mock = build_mock_pollers(mh);
     let core = mock_worker(mock);
-    let mut worker = TestRustWorker::new(Arc::new(core), TEST_Q.to_string(), None);
+    let mut worker = Worker::new(Arc::new(core), TEST_Q.to_string(), None);
 
     worker.register_wf(
         DEFAULT_WORKFLOW_TYPE.to_owned(),
@@ -310,7 +310,7 @@ async fn local_act_retry_long_backoff_uses_timer() {
     let mut mock = build_mock_pollers(mh);
     mock.worker_cfg(|w| w.max_cached_workflows = 1);
     let core = mock_worker(mock);
-    let mut worker = TestRustWorker::new(Arc::new(core), TEST_Q.to_string(), None);
+    let mut worker = Worker::new(Arc::new(core), TEST_Q.to_string(), None);
 
     worker.register_wf(
         DEFAULT_WORKFLOW_TYPE.to_owned(),

sdk/src/lib.rs

@@ -35,7 +35,7 @@ use std::{
 use temporal_client::{ServerGatewayApis, WorkflowOptions};
 use temporal_sdk_core_api::{
     errors::{PollActivityError, PollWfError},
-    Worker,
+    Worker as CoreWorker,
 };
 use temporal_sdk_core_protos::{
     coresdk::{
@@ -64,9 +64,10 @@ use tokio::{
 };
 use tokio_util::sync::CancellationToken;
 
-/// A worker that can poll for and respond to workflow tasks by using [WorkflowFunction]s
-pub struct TestRustWorker {
-    worker: Arc<dyn Worker>,
+/// A worker that can poll for and respond to workflow tasks by using [WorkflowFunction]s,
+/// and activity tasks by using [ActivityFunction]s
+pub struct Worker {
+    worker: Arc<dyn CoreWorker>,
     task_queue: String,
     task_timeout: Option<Duration>,
     workflow_half: WorkflowHalf,
@@ -91,10 +92,10 @@ struct ActivityHalf {
     task_tokens_to_cancels: HashMap<TaskToken, CancellationToken>,
 }
 
-impl TestRustWorker {
+impl Worker {
     /// Create a new rust worker
     pub fn new(
-        worker: Arc<dyn Worker>,
+        worker: Arc<dyn CoreWorker>,
         task_queue: String,
         task_timeout: Option<Duration>,
     ) -> Self {
@@ -280,11 +281,18 @@ impl TestRustWorker {
     /// Turns this rust worker into a new worker with all the same workflows and activities
     /// registered, but with a new underlying core worker. Can be used to swap the worker for
     /// a replay worker, change task queues, etc.
-    pub fn with_new_core_worker(&mut self, new_core_worker: Arc<dyn Worker>) {
+    pub fn with_new_core_worker(&mut self, new_core_worker: Arc<dyn CoreWorker>) {
         self.worker = new_core_worker;
     }
 
-    fn split_apart(&mut self) -> (Arc<dyn Worker>, &str, &mut WorkflowHalf, &mut ActivityHalf) {
+    fn split_apart(
+        &mut self,
+    ) -> (
+        Arc<dyn CoreWorker>,
+        &str,
+        &mut WorkflowHalf,
+        &mut ActivityHalf,
+    ) {
         (
             self.worker.clone(),
             &self.task_queue,
@@ -297,7 +305,7 @@ impl TestRustWorker {
 impl WorkflowHalf {
     async fn workflow_activation_handler(
         &mut self,
-        worker: &dyn Worker,
+        worker: &dyn CoreWorker,
         task_queue: &str,
         shutdown_rx: &Receiver<bool>,
         completions_tx: &UnboundedSender<WorkflowActivationCompletion>,
@@ -374,7 +382,7 @@ impl ActivityHalf {
     /// Spawns off a task to handle the provided activity task
     fn activity_task_handler(
         &mut self,
-        worker: Arc<dyn Worker>,
+        worker: Arc<dyn CoreWorker>,
         activity: ActivityTask,
     ) -> Result<(), anyhow::Error> {
         match activity.variant {

test-utils/src/lib.rs

@@ -12,13 +12,13 @@ use std::{
     time::Duration,
 };
 use temporal_client::WorkflowOptions;
-use temporal_sdk::TestRustWorker;
+use temporal_sdk::Worker;
 use temporal_sdk_core::{
     init_replay_worker, init_worker, replay::mock_gateway_from_history, telemetry_init,
     ServerGatewayOptions, ServerGatewayOptionsBuilder, TelemetryOptions, TelemetryOptionsBuilder,
     WorkerConfig, WorkerConfigBuilder,
 };
-use temporal_sdk_core_api::Worker;
+use temporal_sdk_core_api::Worker as CoreWorker;
 use temporal_sdk_core_protos::{
     coresdk::{
         workflow_commands::{
@@ -40,7 +40,7 @@ const PROM_ENABLE_ENV_VAR: &str = "TEMPORAL_INTEG_PROM_PORT";
 
 /// Create a worker instance which will use the provided test name to base the task queue and wf id
 /// upon. Returns the instance and the task queue name (which is also the workflow id).
-pub async fn init_core_and_create_wf(test_name: &str) -> (Arc<dyn Worker>, String) {
+pub async fn init_core_and_create_wf(test_name: &str) -> (Arc<dyn CoreWorker>, String) {
     let mut starter = CoreWfStarter::new(test_name);
     let core = starter.get_worker().await;
     starter.start_wf().await;
@@ -49,7 +49,10 @@ pub async fn init_core_and_create_wf(test_name: &str) -> (Arc<dyn Worker>, Strin
 
 /// Create a worker replay instance preloaded with a provided history. Returns the worker impl
 /// and the task queue name as in [init_core_and_create_wf].
-pub fn init_core_replay_preloaded(test_name: &str, history: &History) -> (Arc<dyn Worker>, String) {
+pub fn init_core_replay_preloaded(
+    test_name: &str,
+    history: &History,
+) -> (Arc<dyn CoreWorker>, String) {
     let worker_cfg = WorkerConfigBuilder::default()
         .namespace(NAMESPACE)
         .task_queue(test_name)
@@ -77,7 +80,7 @@ pub struct CoreWfStarter {
     telemetry_options: TelemetryOptions,
     worker_config: WorkerConfig,
     wft_timeout: Option<Duration>,
-    initted_worker: Option<Arc<dyn Worker>>,
+    initted_worker: Option<Arc<dyn CoreWorker>>,
 }
 
 impl CoreWfStarter {
@@ -103,8 +106,8 @@ impl CoreWfStarter {
         }
     }
 
-    pub async fn worker(&mut self) -> TestRustWorker {
-        TestRustWorker::new(
+    pub async fn worker(&mut self) -> Worker {
+        Worker::new(
             self.get_worker().await,
             self.worker_config.task_queue.clone(),
             self.wft_timeout,
@@ -115,7 +118,7 @@ impl CoreWfStarter {
         self.get_worker().await.shutdown().await;
     }
 
-    pub async fn get_worker(&mut self) -> Arc<dyn Worker> {
+    pub async fn get_worker(&mut self) -> Arc<dyn CoreWorker> {
         if self.initted_worker.is_none() {
             telemetry_init(&self.telemetry_options).expect("Telemetry inits cleanly");
             let gateway = get_integ_server_options()
@@ -163,7 +166,7 @@ impl CoreWfStarter {
         wf_id: impl Into<String>,
         run_id: impl Into<String>,
         // TODO: Need not be passed in
-        worker: &mut TestRustWorker,
+        worker: &mut Worker,
     ) -> Result<(), anyhow::Error> {
         // Fetch history and replay it
         let history = self
@@ -317,7 +320,7 @@ pub trait WorkerTestHelpers {
 #[async_trait::async_trait]
 impl<T> WorkerTestHelpers for T
 where
-    T: Worker + ?Sized,
+    T: CoreWorker + ?Sized,
 {
     async fn complete_execution(&self, run_id: &str) {
         self.complete_workflow_activation(WorkflowActivationCompletion::from_cmds(

tests/integ_tests/workflow_tests/replay.rs

@@ -1,6 +1,6 @@
 use assert_matches::assert_matches;
 use std::time::Duration;
-use temporal_sdk::{TestRustWorker, WfContext, WorkflowFunction};
+use temporal_sdk::{WfContext, Worker, WorkflowFunction};
 use temporal_sdk_core::{replay::mock_gateway_from_history, telemetry_init, ServerGatewayApis};
 use temporal_sdk_core_api::errors::{PollActivityError, PollWfError};
 use temporal_sdk_core_protos::{
@@ -133,7 +133,7 @@ async fn replay_using_wf_function() {
     let func = timers_wf(num_timers);
     let (worker, _) =
         init_core_replay_preloaded("replay_bench", &t.get_full_history_info().unwrap().into());
-    let mut worker = TestRustWorker::new(worker, "replay_bench".to_string(), None);
+    let mut worker = Worker::new(worker, "replay_bench".to_string(), None);
     worker.register_wf(DEFAULT_WORKFLOW_TYPE, func);
     worker.incr_expected_run_count(1);
     worker.run_until_done().await.unwrap();