Support separate activity/workflow workers

Author: cschleiden

internal/worker/activity.go

@@ -34,7 +34,7 @@ func NewActivityWorker(
 		logger:               b.Options().Logger,
 	}
 
-	return NewWorker[backend.ActivityTask, history.Event](b, tw, &options)
+	return NewWorker(b, tw, &options)
 }
 
 type ActivityTaskWorker struct {

samples/scale/worker/main.go

@@ -33,10 +33,14 @@ func main() {
 
 func RunWorker(ctx context.Context, mb backend.Backend) {
 	w := worker.New(mb, &worker.Options{
-		WorkflowPollers:          1,
-		MaxParallelWorkflowTasks: 100,
-		ActivityPollers:          1,
-		MaxParallelActivityTasks: 100,
+		WorkflowWorkerOptions: worker.WorkflowWorkerOptions{
+			WorkflowPollers:          1,
+			MaxParallelWorkflowTasks: 100,
+		},
+		ActivityWorkerOptions: worker.ActivityWorkerOptions{
+			ActivityPollers:          1,
+			MaxParallelActivityTasks: 100,
+		},
 	})
 
 	w.RegisterWorkflow(scale.Workflow1)

worker/options.go

@@ -7,25 +7,14 @@ import (
 	"github.com/cschleiden/go-workflows/workflow/executor"
 )
 
-type Options struct {
+type WorkflowWorkerOptions struct {
 	// WorkflowsPollers is the number of pollers to start. Defaults to 2.
 	WorkflowPollers int
 
 	// MaxParallelWorkflowTasks determines the maximum number of concurrent workflow tasks processed
 	// by the worker. The default is 0 which is no limit.
 	MaxParallelWorkflowTasks int
 
-	// ActivityPollers is the number of pollers to start. Defaults to 2.
-	ActivityPollers int
-
-	// MaxParallelActivityTasks determines the maximum number of concurrent activity tasks processed
-	// 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
-
 	// WorkflowHeartbeatInterval is the interval between heartbeat attempts on workflow tasks. Defaults
 	// to 25 seconds
 	WorkflowHeartbeatInterval time.Duration
@@ -35,11 +24,6 @@ type Options struct {
 	// Defaults to 200ms.
 	WorkflowPollingInterval time.Duration
 
-	// ActivityPollingInterval is the interval between polling for new activity tasks.
-	// Note that if you use a backend that can wait for tasks to be available (e.g. redis) this field has no effect.
-	// Defaults to 200ms.
-	ActivityPollingInterval time.Duration
-
 	// WorkflowExecutorCache is the max size of the workflow executor cache. Defaults to 128
 	WorkflowExecutorCacheSize int
 
@@ -50,23 +34,52 @@ type Options struct {
 	// will be used.
 	WorkflowExecutorCache executor.Cache
 
+	// WorkflowQueues are the queue the worker listens to
 	WorkflowQueues []workflow.Queue
+}
+
+type Options struct {
+	WorkflowWorkerOptions
+	ActivityWorkerOptions
+}
+
+type ActivityWorkerOptions struct {
+	// ActivityPollers is the number of pollers to start. Defaults to 2.
+	ActivityPollers int
+
+	// MaxParallelActivityTasks determines the maximum number of concurrent activity tasks processed
+	// 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
+
+	// ActivityPollingInterval is the interval between polling for new activity tasks.
+	// Note that if you use a backend that can wait for tasks to be available (e.g. redis) this field has no effect.
+	// Defaults to 200ms.
+	ActivityPollingInterval time.Duration
 
+	// ActivityQueues are the queues the worker listens to
 	ActivityQueues []workflow.Queue
 }
 
 var DefaultOptions = Options{
-	WorkflowPollers:           2,
-	WorkflowPollingInterval:   200 * time.Millisecond,
-	MaxParallelWorkflowTasks:  0,
-	WorkflowHeartbeatInterval: 25 * time.Second,
-
-	WorkflowExecutorCacheSize: 128,
-	WorkflowExecutorCacheTTL:  time.Second * 10,
-	WorkflowExecutorCache:     nil,
-
-	ActivityPollers:           2,
-	ActivityPollingInterval:   200 * time.Millisecond,
-	MaxParallelActivityTasks:  0,
-	ActivityHeartbeatInterval: 25 * time.Second,
+	WorkflowWorkerOptions: WorkflowWorkerOptions{
+		WorkflowPollers:           2,
+		WorkflowPollingInterval:   200 * time.Millisecond,
+		MaxParallelWorkflowTasks:  0,
+		WorkflowHeartbeatInterval: 25 * time.Second,
+
+		WorkflowExecutorCacheSize: 128,
+		WorkflowExecutorCacheTTL:  time.Second * 10,
+		WorkflowExecutorCache:     nil,
+	},
+
+	ActivityWorkerOptions: ActivityWorkerOptions{
+		ActivityPollers:           2,
+		ActivityPollingInterval:   200 * time.Millisecond,
+		MaxParallelActivityTasks:  0,
+		ActivityHeartbeatInterval: 25 * time.Second,
+	},
 }

worker/worker.go

@@ -3,106 +3,128 @@ package worker
 import (
 	"context"
 	"fmt"
-	"sync"
 
 	"github.com/benbjohnson/clock"
 	"github.com/cschleiden/go-workflows/backend"
-	"github.com/cschleiden/go-workflows/backend/history"
 	"github.com/cschleiden/go-workflows/client"
 	"github.com/cschleiden/go-workflows/internal/signals"
 	internal "github.com/cschleiden/go-workflows/internal/worker"
 	"github.com/cschleiden/go-workflows/registry"
 	"github.com/cschleiden/go-workflows/workflow"
-	"github.com/cschleiden/go-workflows/workflow/executor"
 )
 
 type Worker struct {
 	backend backend.Backend
 
-	done chan struct{}
-	wg   *sync.WaitGroup
-
 	registry *registry.Registry
 
-	workflowWorker *internal.Worker[backend.WorkflowTask, executor.ExecutionResult]
-	activityWorker *internal.Worker[backend.ActivityTask, history.Event]
+	workers []worker
+}
+
+type worker interface {
+	Start(context.Context) error
+	WaitForCompletion() error
 }
 
+// New creates a worker that processes workflows and activities.
 func New(backend backend.Backend, options *Options) *Worker {
+	registry := registry.New()
+
 	if options == nil {
 		options = &DefaultOptions
-	} else {
-		if options.WorkflowExecutorCacheSize == 0 {
-			options.WorkflowExecutorCacheSize = DefaultOptions.WorkflowExecutorCacheSize
-		}
-
-		if options.WorkflowExecutorCacheTTL == 0 {
-			options.WorkflowExecutorCacheTTL = DefaultOptions.WorkflowExecutorCacheTTL
-		}
 	}
 
-	registry := registry.New()
+	workflowWorker := newWorkflowWorker(backend, registry, &options.WorkflowWorkerOptions)
+	activityWorker := newActivityWorker(backend, registry, &options.ActivityWorkerOptions)
 
 	// Register internal activities
+	return newWorker(backend, registry, []worker{workflowWorker, activityWorker})
+}
+
+// NewWorkflowWorker creates a worker that only processes workflows.
+func NewWorkflowWorker(backend backend.Backend, options *WorkflowWorkerOptions) *Worker {
+	registry := registry.New()
+
+	return newWorker(backend, registry, []worker{newWorkflowWorker(backend, registry, options)})
+}
+
+// NewActivityWorker creates a worker that only processes activities.
+func NewActivityWorker(backend backend.Backend, options *ActivityWorkerOptions) *Worker {
+	registry := registry.New()
+
+	return newWorker(backend, registry, []worker{newActivityWorker(backend, registry, options)})
+}
+
+func newWorker(backend backend.Backend, registry *registry.Registry, workers []worker) *Worker {
 	if err := registry.RegisterActivity(&signals.Activities{Signaler: client.New(backend)}); err != nil {
 		panic(fmt.Errorf("registering internal activities: %w", err))
 	}
 
 	return &Worker{
 		backend: backend,
 
-		done: make(chan struct{}),
-		wg:   &sync.WaitGroup{},
-
-		workflowWorker: internal.NewWorkflowWorker(backend, registry, internal.WorkflowWorkerOptions{
-			WorkerOptions: internal.WorkerOptions{
-				Pollers:           options.WorkflowPollers,
-				PollingInterval:   options.WorkflowPollingInterval,
-				MaxParallelTasks:  options.MaxParallelWorkflowTasks,
-				HeartbeatInterval: options.WorkflowHeartbeatInterval,
-				Queues:            options.WorkflowQueues,
-			},
-			WorkflowExecutorCache:     options.WorkflowExecutorCache,
-			WorkflowExecutorCacheSize: options.WorkflowExecutorCacheSize,
-			WorkflowExecutorCacheTTL:  options.WorkflowExecutorCacheTTL,
-		}),
-
-		activityWorker: internal.NewActivityWorker(backend, registry, clock.New(), internal.WorkerOptions{
-			Pollers:           options.ActivityPollers,
-			PollingInterval:   options.ActivityPollingInterval,
-			MaxParallelTasks:  options.MaxParallelActivityTasks,
-			HeartbeatInterval: options.ActivityHeartbeatInterval,
-			Queues:            options.ActivityQueues,
-		}),
-
+		workers:  workers,
 		registry: registry,
 	}
 }
 
+func newActivityWorker(backend backend.Backend, registry *registry.Registry, options *ActivityWorkerOptions) worker {
+	if options == nil {
+		options = &DefaultOptions.ActivityWorkerOptions
+	}
+
+	activityWorker := internal.NewActivityWorker(backend, registry, clock.New(), internal.WorkerOptions{
+		Pollers:           options.ActivityPollers,
+		PollingInterval:   options.ActivityPollingInterval,
+		MaxParallelTasks:  options.MaxParallelActivityTasks,
+		HeartbeatInterval: options.ActivityHeartbeatInterval,
+		Queues:            options.ActivityQueues,
+	})
+
+	return activityWorker
+}
+
+func newWorkflowWorker(backend backend.Backend, registry *registry.Registry, options *WorkflowWorkerOptions) worker {
+	if options == nil {
+		options = &DefaultOptions.WorkflowWorkerOptions
+	}
+
+	workflowWorker := internal.NewWorkflowWorker(backend, registry, internal.WorkflowWorkerOptions{
+		WorkerOptions: internal.WorkerOptions{
+			Pollers:           options.WorkflowPollers,
+			PollingInterval:   options.WorkflowPollingInterval,
+			MaxParallelTasks:  options.MaxParallelWorkflowTasks,
+			HeartbeatInterval: options.WorkflowHeartbeatInterval,
+			Queues:            options.WorkflowQueues,
+		},
+		WorkflowExecutorCache:     options.WorkflowExecutorCache,
+		WorkflowExecutorCacheSize: options.WorkflowExecutorCacheSize,
+		WorkflowExecutorCacheTTL:  options.WorkflowExecutorCacheTTL,
+	})
+
+	return workflowWorker
+}
+
 // Start starts the worker.
 //
 // To stop the worker, cancel the context passed to Start. To wait for completion of the active
 // tasks, call `WaitForCompletion`.
 func (w *Worker) Start(ctx context.Context) error {
-	if err := w.workflowWorker.Start(ctx); err != nil {
-		return fmt.Errorf("starting workflow worker: %w", err)
-	}
-
-	if err := w.activityWorker.Start(ctx); err != nil {
-		return fmt.Errorf("starting activity worker: %w", err)
+	for _, worker := range w.workers {
+		if err := worker.Start(ctx); err != nil {
+			return fmt.Errorf("starting worker: %w", err)
+		}
 	}
 
 	return nil
 }
 
 // WaitForCompletion waits for all active tasks to complete.
 func (w *Worker) WaitForCompletion() error {
-	if err := w.workflowWorker.WaitForCompletion(); err != nil {
-		return err
-	}
-
-	if err := w.activityWorker.WaitForCompletion(); err != nil {
-		return err
+	for _, worker := range w.workers {
+		if err := worker.WaitForCompletion(); err != nil {
+			return fmt.Errorf("waiting for worker completion: %w", err)
+		}
 	}
 
 	return nil