Merge pull request #351 from cschleiden/queues

Add support for workflow and activity queues

Author: cschleiden

backend/backend.go

@@ -3,9 +3,7 @@ package backend
 import (
 	"context"
 	"errors"
-	"log/slog"
 
-	"github.com/cschleiden/go-workflows/backend/converter"
 	"github.com/cschleiden/go-workflows/backend/history"
 	"github.com/cschleiden/go-workflows/backend/metrics"
 	"github.com/cschleiden/go-workflows/core"
@@ -42,47 +40,47 @@ type Backend interface {
 	// If the given instance does not exist, it will return an error
 	SignalWorkflow(ctx context.Context, instanceID string, event *history.Event) error
 
+	// PrepareWorkflowQueues prepares workflow queues for later consumption using this backend instane
+	PrepareWorkflowQueues(ctx context.Context, queues []workflow.Queue) error
+
+	// PrepareActivityQueues prepares activity queues for later consumption using this backend instance
+	PrepareActivityQueues(ctx context.Context, queues []workflow.Queue) error
+
 	// GetWorkflowTask returns a pending workflow task or nil if there are no pending workflow executions
-	GetWorkflowTask(ctx context.Context) (*WorkflowTask, error)
+	GetWorkflowTask(ctx context.Context, queues []workflow.Queue) (*WorkflowTask, error)
 
 	// ExtendWorkflowTask extends the lock of a workflow task
-	ExtendWorkflowTask(ctx context.Context, taskID string, instance *core.WorkflowInstance) error
+	ExtendWorkflowTask(ctx context.Context, task *WorkflowTask) error
 
 	// CompleteWorkflowTask checkpoints a workflow task retrieved using GetWorkflowTask
 	//
 	// This checkpoints the execution. events are new events from the last workflow execution
 	// which will be added to the workflow instance history. workflowEvents are new events for the
 	// completed or other workflow instances.
 	CompleteWorkflowTask(
-		ctx context.Context, task *WorkflowTask, instance *workflow.Instance, state core.WorkflowInstanceState,
+		ctx context.Context, task *WorkflowTask, state core.WorkflowInstanceState,
 		executedEvents, activityEvents, timerEvents []*history.Event, workflowEvents []*history.WorkflowEvent) error
 
 	// GetActivityTask returns a pending activity task or nil if there are no pending activities
-	GetActivityTask(ctx context.Context) (*ActivityTask, error)
-
-	// CompleteActivityTask completes an activity task retrieved using GetActivityTask
-	CompleteActivityTask(ctx context.Context, instance *workflow.Instance, activityID string, event *history.Event) error
+	GetActivityTask(ctx context.Context, queues []workflow.Queue) (*ActivityTask, error)
 
 	// ExtendActivityTask extends the lock of an activity task
-	ExtendActivityTask(ctx context.Context, activityID string) error
+	ExtendActivityTask(ctx context.Context, task *ActivityTask) error
+
+	// CompleteActivityTask completes an activity task retrieved using GetActivityTask
+	CompleteActivityTask(ctx context.Context, task *ActivityTask, result *history.Event) error
 
 	// GetStats returns stats about the backend
 	GetStats(ctx context.Context) (*Stats, error)
 
-	// Logger returns the configured logger for the backend
-	Logger() *slog.Logger
-
 	// Tracer returns the configured trace provider for the backend
 	Tracer() trace.Tracer
 
 	// Metrics returns the configured metrics client for the backend
 	Metrics() metrics.Client
 
-	// Converter returns the configured converter for the backend
-	Converter() converter.Converter
-
-	// ContextPropagators returns the configured context propagators for the backend
-	ContextPropagators() []workflow.ContextPropagator
+	// Options returns the configured options for the backend
+	Options() *Options
 
 	// Close closes any underlying resources
 	Close() error

backend/history/activity_scheduled.go

@@ -3,6 +3,7 @@ package history
 import (
 	"github.com/cschleiden/go-workflows/backend/metadata"
 	"github.com/cschleiden/go-workflows/backend/payload"
+	"github.com/cschleiden/go-workflows/core"
 )
 
 type ActivityScheduledAttributes struct {
@@ -13,4 +14,6 @@ type ActivityScheduledAttributes struct {
 	Inputs []payload.Payload `json:"inputs,omitempty"`
 
 	Metadata *metadata.WorkflowMetadata `json:"metadata,omitempty"`
+
+	Queue core.Queue `json:"queue,omitempty"`
 }

backend/history/subworkflow_scheduled.go

@@ -7,6 +7,8 @@ import (
 )
 
 type SubWorkflowScheduledAttributes struct {
+	SubWorkflowQueue core.Queue `json:"sub_workflow_queue,omitempty"`
+
 	SubWorkflowInstance *core.WorkflowInstance `json:"sub_workflow_instance,omitempty"`
 
 	Name string `json:"name,omitempty"`

backend/history/workflow_started.go

@@ -3,9 +3,12 @@ package history
 import (
 	"github.com/cschleiden/go-workflows/backend/metadata"
 	"github.com/cschleiden/go-workflows/backend/payload"
+	"github.com/cschleiden/go-workflows/core"
 )
 
 type ExecutionStartedAttributes struct {
+	Queue core.Queue `json:"queue,omitempty"`
+
 	Name string `json:"name,omitempty"`
 
 	Metadata *metadata.WorkflowMetadata `json:"metadata,omitempty"`