feat(go): migrated activity timeouts

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

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
+}

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())
+}

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
+}