diff --git a/go-sandbox/activities/timeouts/activity.go b/go-sandbox/activities/timeouts/activity.go new file mode 100644 index 0000000..2aaccbd --- /dev/null +++ b/go-sandbox/activities/timeouts/activity.go @@ -0,0 +1,64 @@ +package timeouts + +import ( + "context" + "fmt" + "time" + + "go.temporal.io/sdk/activity" + "go.temporal.io/sdk/temporal" +) + +type YourActivityParam struct { + Items []string +} + +type YourActivityResult struct { + ProcessedCount int +} + +// @@@SNIPSTART go-activity-heartbeat +func YourActivityDefinition(ctx context.Context, param YourActivityParam) (YourActivityResult, error) { + for i := range param.Items { + // Process item i... + details := fmt.Sprintf("processed item %d", i) + activity.RecordHeartbeat(ctx, details) + } + return YourActivityResult{ProcessedCount: len(param.Items)}, nil +} +// @@@SNIPEND + +func NextRetryDelayActivity(ctx context.Context) error { + // @@@SNIPSTART go-activity-next-retry-delay + attempt := activity.GetInfo(ctx).Attempt + + return temporal.NewApplicationErrorWithOptions(fmt.Sprintf("Something bad happened on attempt %d", attempt), "NextDelay", temporal.ApplicationErrorOptions{ + NextRetryDelay: 3 * time.Second * time.Duration(attempt), + }) + // @@@SNIPEND +} + +type InputParams struct { + StartIndex int + EndIndex int +} + +// @@@SNIPSTART go-activity-heartbeat-resume +func SampleActivity(ctx context.Context, inputArg InputParams) error { + startIdx := inputArg.StartIndex + if activity.HasHeartbeatDetails(ctx) { + // Recover from finished progress. + var finishedIndex int + if err := activity.GetHeartbeatDetails(ctx, &finishedIndex); err == nil { + startIdx = finishedIndex + 1 // Start from next one. + } + } + + // Normal Activity logic... + for i := startIdx; i < inputArg.EndIndex; i++ { + // Code for processing item i goes here... + activity.RecordHeartbeat(ctx, i) // Report progress. + } + return nil +} +// @@@SNIPEND diff --git a/go-sandbox/activities/timeouts/heartbeat.go b/go-sandbox/activities/timeouts/heartbeat.go new file mode 100644 index 0000000..7761fab --- /dev/null +++ b/go-sandbox/activities/timeouts/heartbeat.go @@ -0,0 +1,23 @@ +package timeouts + +import ( + "context" + + "go.temporal.io/sdk/client" +) + +// RecordExternalHeartbeat Heartbeats an Activity from outside the Worker, using +// the Task Token retrieved inside the Activity. +func RecordExternalHeartbeat(ctx context.Context, taskToken []byte, details string) error { + // @@@SNIPSTART go-activity-heartbeat-external + // The client is a heavyweight object that should be created once per process. + temporalClient, err := client.Dial(client.Options{}) + if err != nil { + return err + } + defer temporalClient.Close() + + // Record heartbeat. + return temporalClient.RecordActivityHeartbeat(ctx, taskToken, details) + // @@@SNIPEND +} diff --git a/go-sandbox/activities/timeouts/timeouts_test.go b/go-sandbox/activities/timeouts/timeouts_test.go new file mode 100644 index 0000000..01998ab --- /dev/null +++ b/go-sandbox/activities/timeouts/timeouts_test.go @@ -0,0 +1,86 @@ +package timeouts_test + +import ( + "errors" + "testing" + "time" + + "github.com/stretchr/testify/suite" + "go.temporal.io/sdk/activity" + "go.temporal.io/sdk/converter" + "go.temporal.io/sdk/temporal" + "go.temporal.io/sdk/testsuite" + + "my-org/quickstart/activities/timeouts" +) + +type TimeoutsTestSuite struct { + suite.Suite + testsuite.WorkflowTestSuite +} + +func TestTimeoutsTestSuite(t *testing.T) { + suite.Run(t, new(TimeoutsTestSuite)) +} + +func (s *TimeoutsTestSuite) Test_Workflows() { + param := timeouts.YourActivityParam{Items: []string{"a", "b", "c"}} + for _, wf := range []interface{}{ + timeouts.ActivityTimeoutWorkflow, + timeouts.CustomRetryPolicyWorkflow, + timeouts.HeartbeatTimeoutWorkflow, + } { + env := s.NewTestWorkflowEnvironment() + env.RegisterActivity(timeouts.YourActivityDefinition) + env.ExecuteWorkflow(wf, param) + s.True(env.IsWorkflowCompleted()) + s.NoError(env.GetWorkflowError()) + + var result timeouts.YourActivityResult + s.NoError(env.GetWorkflowResult(&result)) + s.Equal(3, result.ProcessedCount) + } +} + +func (s *TimeoutsTestSuite) Test_YourActivityDefinitionHeartbeats() { + env := s.NewTestActivityEnvironment() + env.RegisterActivity(timeouts.YourActivityDefinition) + var heartbeats int + env.SetOnActivityHeartbeatListener(func(info *activity.Info, details converter.EncodedValues) { + heartbeats++ + }) + + _, err := env.ExecuteActivity(timeouts.YourActivityDefinition, timeouts.YourActivityParam{Items: []string{"a", "b"}}) + s.NoError(err) + // The Worker throttles Heartbeats, so only the first one is guaranteed to be sent. + s.GreaterOrEqual(heartbeats, 1) +} + +func (s *TimeoutsTestSuite) Test_SampleActivityResumesFromHeartbeat() { + env := s.NewTestActivityEnvironment() + env.RegisterActivity(timeouts.SampleActivity) + env.SetHeartbeatDetails(4) + var reported []int + env.SetOnActivityHeartbeatListener(func(info *activity.Info, details converter.EncodedValues) { + var i int + s.NoError(details.Get(&i)) + reported = append(reported, i) + }) + + _, err := env.ExecuteActivity(timeouts.SampleActivity, timeouts.InputParams{StartIndex: 0, EndIndex: 7}) + s.NoError(err) + // The Activity resumes after the index in the last Heartbeat details. + s.NotEmpty(reported) + s.Equal(5, reported[0]) +} + +func (s *TimeoutsTestSuite) Test_NextRetryDelay() { + env := s.NewTestActivityEnvironment() + env.RegisterActivity(timeouts.NextRetryDelayActivity) + + _, err := env.ExecuteActivity(timeouts.NextRetryDelayActivity) + var appErr *temporal.ApplicationError + s.True(errors.As(err, &appErr)) + s.Equal("NextDelay", appErr.Type()) + s.Equal(3*time.Second, appErr.NextRetryDelay()) +} diff --git a/go-sandbox/activities/timeouts/workflow.go b/go-sandbox/activities/timeouts/workflow.go new file mode 100644 index 0000000..c945691 --- /dev/null +++ b/go-sandbox/activities/timeouts/workflow.go @@ -0,0 +1,64 @@ +package timeouts + +import ( + "time" + + "go.temporal.io/sdk/temporal" + "go.temporal.io/sdk/workflow" +) + +func ActivityTimeoutWorkflow(ctx workflow.Context, yourActivityParam YourActivityParam) (*YourActivityResult, error) { + // @@@SNIPSTART go-activity-timeouts-set + activityOptions := workflow.ActivityOptions{ + // Set Activity Timeout duration + ScheduleToCloseTimeout: 10 * time.Second, + // StartToCloseTimeout: 10 * time.Second, + // ScheduleToStartTimeout: 10 * time.Second, + } + ctx = workflow.WithActivityOptions(ctx, activityOptions) + var yourActivityResult YourActivityResult + err := workflow.ExecuteActivity(ctx, YourActivityDefinition, yourActivityParam).Get(ctx, &yourActivityResult) + if err != nil { + return nil, err + } + // @@@SNIPEND + return &yourActivityResult, nil +} + +func CustomRetryPolicyWorkflow(ctx workflow.Context, yourActivityParam YourActivityParam) (*YourActivityResult, error) { + // @@@SNIPSTART go-activity-retry-policy-custom + retryPolicy := &temporal.RetryPolicy{ + InitialInterval: time.Second, + BackoffCoefficient: 2.0, + MaximumInterval: time.Second * 100, + } + + activityOptions := workflow.ActivityOptions{ + StartToCloseTimeout: 10 * time.Second, + RetryPolicy: retryPolicy, + } + ctx = workflow.WithActivityOptions(ctx, activityOptions) + var yourActivityResult YourActivityResult + err := workflow.ExecuteActivity(ctx, YourActivityDefinition, yourActivityParam).Get(ctx, &yourActivityResult) + if err != nil { + return nil, err + } + // @@@SNIPEND + return &yourActivityResult, nil +} + +func HeartbeatTimeoutWorkflow(ctx workflow.Context, yourActivityParam YourActivityParam) (*YourActivityResult, error) { + // @@@SNIPSTART go-activity-heartbeat-timeout + activityOptions := workflow.ActivityOptions{ + StartToCloseTimeout: 10 * time.Minute, + HeartbeatTimeout: 10 * time.Second, + } + ctx = workflow.WithActivityOptions(ctx, activityOptions) + var yourActivityResult YourActivityResult + err := workflow.ExecuteActivity(ctx, YourActivityDefinition, yourActivityParam).Get(ctx, &yourActivityResult) + if err != nil { + return nil, err + } + // @@@SNIPEND + return &yourActivityResult, nil +}