Merge pull request #97 from cschleiden/heartbeat-configuration

Make heartbeat interval configurable

Author: cschleiden

internal/worker/activity.go

@@ -121,7 +121,7 @@ func (aw *activityWorker) handleTask(ctx context.Context, task *task.Activity) {
 	heartbeatCtx, cancelHeartbeat := context.WithCancel(ctx)
 
 	go func(ctx context.Context) {
-		t := time.NewTicker(30 * time.Second)
+		t := time.NewTicker(aw.options.ActivityHeartbeatInterval)
 		defer t.Stop()
 
 		for {

internal/worker/options.go

@@ -1,5 +1,7 @@
 package worker
 
+import "time"
+
 type Options struct {
 	// WorkflowsPollers is the number of pollers to start. Defaults to 2.
 	WorkflowPollers int
@@ -15,15 +17,24 @@ type Options struct {
 	// by the worker. The default is 0 which is no limit.
 	MaxParallelActivityTasks int
 
+	// ActivityHeartbeatInterval is the interval between heartbeat attempts for activity tasks. Defaults
+	// to 25 seconds
+	ActivityHeartbeatInterval time.Duration
+
 	// HeartbeatWorkflowTasks determines if the lock on workflow tasks should be periodically
 	// extended while they are being processed. Given that workflow executions should be
 	// very quick, this is usually not necessary.
 	HeartbeatWorkflowTasks bool
+
+	// WorkflowHeartbeatInterval is the interval between heartbeat attempts on workflow tasks, when enabled.
+	WorkflowHeartbeatInterval time.Duration
 }
 
 var DefaultOptions = Options{
-	WorkflowPollers:          2,
-	ActivityPollers:          2,
-	MaxParallelWorkflowTasks: 0,
-	MaxParallelActivityTasks: 0,
+	WorkflowPollers:           2,
+	ActivityPollers:           2,
+	MaxParallelWorkflowTasks:  0,
+	MaxParallelActivityTasks:  0,
+	ActivityHeartbeatInterval: 25 * time.Second,
+	WorkflowHeartbeatInterval: 25 * time.Second,
 }

internal/worker/workflow.go

@@ -185,8 +185,7 @@ func (ww *workflowWorker) getExecutor(ctx context.Context, t *task.Workflow) (wo
 }
 
 func (ww *workflowWorker) heartbeatTask(ctx context.Context, task *task.Workflow) {
-	// TODO: Make configurable
-	t := time.NewTicker(25 * time.Second)
+	t := time.NewTicker(ww.options.WorkflowHeartbeatInterval)
 	defer t.Stop()
 
 	for {