diff --git a/README.md b/README.md index 6de298e..ca150a5 100644 --- a/README.md +++ b/README.md @@ -168,6 +168,7 @@ Each step can have the following fields: - `c`: (array of steps, optional) Child steps to execute as a child workflow - `r`: (int, optional) Number of times to repeat this step (default 1) - `p`: (int, optional) Size in bytes of padding data to add to activity inputs for increasing history size +- `t`: (int, optional) Seconds to sleep via a durable timer (`workflow.Sleep`) instead of the `Sleep` activity. A value of 0 (the default) is a no-op. #### Examples @@ -194,6 +195,16 @@ This example demonstrates using padding to increase history size by adding paddi ] ``` +This example sleeps for 1 second using a durable timer (`workflow.Sleep`, no activity), runs `Echo`, then sleeps 5 seconds: + +``` +[ + {"t": 1}, + {"a": "Echo", "i": {"Message": "test"}}, + {"t": 5} +] +``` + You can start this workflow using `tctl` or any Temporal client, for example: ``` diff --git a/workflows/workflow.go b/workflows/workflow.go index 845686f..e44c35f 100644 --- a/workflows/workflow.go +++ b/workflows/workflow.go @@ -17,13 +17,15 @@ type ReceiveSignalWorkflowInput struct { Name string } -// DSL step: either an activity or a child workflow (which is always this workflow) +// DSL step: an activity, a child workflow (which is always this workflow), +// and/or a user timer. type DSLStep struct { - Activity string `json:"a,omitempty"` - Input interface{} `json:"i,omitempty"` - Child []DSLStep `json:"c,omitempty"` - Repeat int `json:"r,omitempty"` - PaddingSize int `json:"p,omitempty"` // Size in bytes of padding to add to activity inputs + Activity string `json:"a,omitempty"` + Input interface{} `json:"i,omitempty"` + Child []DSLStep `json:"c,omitempty"` + Repeat int `json:"r,omitempty"` + PaddingSize int `json:"p,omitempty"` // Size in bytes of padding to add to activity inputs + SleepSeconds int `json:"t,omitempty"` // Seconds to sleep using workflow.Sleep } // injectPadding adds padding data to an activity input by adding a Padding field @@ -85,6 +87,11 @@ func DSLWorkflow(ctx workflow.Context, steps []DSLStep) error { repeat = 1 } for i := 0; i < repeat; i++ { + if step.SleepSeconds > 0 { + if err := workflow.Sleep(ctx, time.Duration(step.SleepSeconds)*time.Second); err != nil { + return err + } + } if step.Activity != "" { // Inject padding into the activity input if specified injectPadding(step.Input, step.PaddingSize) diff --git a/workflows/workflow_test.go b/workflows/workflow_test.go index e14675b..133139f 100644 --- a/workflows/workflow_test.go +++ b/workflows/workflow_test.go @@ -3,6 +3,7 @@ package workflows import ( "context" "testing" + "time" "github.com/temporalio/benchmark-workers/activities" @@ -37,6 +38,33 @@ func TestDSLWorkflow(t *testing.T) { require.Equal(t, 6, echoCount, "Echo activity should be called 6 times") } +func TestDSLWorkflowWithTimerSleep(t *testing.T) { + ts := &testsuite.WorkflowTestSuite{} + env := ts.NewTestWorkflowEnvironment() + + env.RegisterActivityWithOptions(activities.EchoActivity, activity.RegisterOptions{Name: "Echo"}) + var echoCount int + env.OnActivity("Echo", mock.Anything, mock.Anything).Return(func(ctx context.Context, input activities.EchoActivityInput) (string, error) { + echoCount++ + return input.Message, nil + }) + + steps := []DSLStep{ + {SleepSeconds: 2, Repeat: 3}, + {Activity: "Echo", Input: map[string]interface{}{"Message": "test"}}, + } + + startTime := env.Now() + env.ExecuteWorkflow(DSLWorkflow, steps) + + require.True(t, env.IsWorkflowCompleted()) + require.NoError(t, env.GetWorkflowError()) + require.Equal(t, 1, echoCount, "Echo activity should be called once") + // The timer sleeps run via workflow.Sleep (durable timers) and are + // fast-forwarded by the test env's time skipping: 3 repeats * 2s. + require.Equal(t, 6*time.Second, env.Now().Sub(startTime)) +} + func TestDSLWorkflowWithPadding(t *testing.T) { ts := &testsuite.WorkflowTestSuite{} env := ts.NewTestWorkflowEnvironment()