Merge pull request #383 from cschleiden/inline

Add single worker mode

Author: cschleiden

.vscode/settings.json

@@ -8,17 +8,10 @@
     "**/CVS": true,
     "**/.DS_Store": true,
     "**/Thumbs.db": true,
-    "**/vendor": true,
+    "**/vendor": true
   },
-  "github-actions.workflows.pinned.workflows": [
-    ".github/workflows/go.yml"
-  ],
+  "github-actions.workflows.pinned.workflows": [],
   "go.testExplorer.showDynamicSubtestsInEditor": true,
   "go.lintTool": "golangci-lint",
-  "Lua.diagnostics.globals": [
-    "KEYS",
-    "ARGV",
-    "redis",
-    "cjson"
-  ]
+  "Lua.diagnostics.globals": ["KEYS", "ARGV", "redis", "cjson"]
 }

docs/source/index.html.md

@@ -58,7 +58,7 @@ func Workflow1(ctx workflow.Context, input string) error {
 }
 ```
 
-Let's first write a simple workflows. Our workflow executes two _activities_ in sequence waiting for each result. Both workflows and activities are written in plain Go. Workflows can be long-running and have to be deterministic so that they can be interrupted and resumed. Activities are functions that can have side-effects and don't have to be deterministic.
+Let's first write a simple workflow. Our workflow executes two _activities_ in sequence waiting for each result. Both workflows and activities are written in plain Go. Workflows can be long-running and have to be deterministic so that they can be interrupted and resumed. Activities are functions that can have side-effects and don't have to be deterministic.
 
 Both workflows and activities support arbitrary inputs and outputs as long as those are serializable.
 
@@ -135,3 +135,71 @@ func main() {
 To finish the example, we create the backend, start a worker in a separate go-routine. We then create a `Client` instance which we then  use to create a new _workflow instance_. A workflow instance is just one running instance of a previously registered workflow.
 
 With the exception of the in-memory backend, we do not have to start the workflow from the same process the worker runs in, we could create the client from another process and create/wait for/cancel/... workflow instances from there.
+
+## Using the WorkflowOrchestrator
+
+For simpler scenarios where you don't need the separation between client and worker, you can use the `WorkflowOrchestrator` which combines both:
+
+```go
+// Define the workflow with activities that will be auto-registered
+func AutoRegisteredWorkflow(ctx workflow.Context, input string) (string, error) {
+	// When using WorkflowOrchestrator, activities are automatically registered
+	// You can directly use any activity function without registering it first
+	r1, err := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity1).Get(ctx)
+	if err != nil {
+		return "", err
+	}
+
+	// Another activity that will be auto-registered
+	r2, err := workflow.ExecuteActivity[string](ctx, workflow.DefaultActivityOptions, FormatResult, input, r1).Get(ctx)
+	if err != nil {
+		return "", err
+	}
+
+	return r2, nil
+}
+
+// Activities are automatically registered when using WorkflowOrchestrator
+func Activity1(ctx context.Context) (int, error) {
+	return 42, nil
+}
+
+func FormatResult(ctx context.Context, input string, value int) (string, error) {
+	return fmt.Sprintf("Processed %s with result %d", input, value), nil
+}
+
+func main() {
+	ctx := context.Background()
+
+	b := sqlite.NewSqliteBackend("simple.sqlite")
+
+	// Create orchestrator instead of separate client and worker
+	orchestrator := worker.NewWorkflowOrchestrator(b, nil)
+
+	// Workflows are automatically registered
+	// Activities defined with InlineActivity don't need registration
+
+	// Start the orchestrator
+	if err := orchestrator.Start(ctx); err != nil {
+		panic("could not start orchestrator")
+	}
+
+	// Create workflow instance
+	wf, err := orchestrator.CreateWorkflowInstance(ctx, client.WorkflowInstanceOptions{
+		InstanceID: uuid.NewString(),
+	}, AutoRegisteredWorkflow, "input-for-workflow")
+	if err != nil {
+		panic("could not start workflow")
+	}
+
+	// Get result directly
+	result, err := client.GetWorkflowResult[string](ctx, orchestrator.Client, wf, 5*time.Second)
+	if err != nil {
+		panic("error getting workflow result: " + err.Error())
+	}
+
+	fmt.Println("Workflow completed with result:", result)
+}
+```
+
+The `WorkflowOrchestrator` provides automatic registration of workflows when passing workflow functions directly to `CreateWorkflowInstance`. For activities, the orchestrator automatically registers any activities used in the workflows via `ExecuteActivity` without explicit registration. Similarly, any sub-workflows created with `CreateSubWorkflowInstance` are registered automatically. This approach offers a unified API for workflow creation and execution in a single component, simplifying the development experience for simpler scenarios.

internal/contextvalue/registry.go

@@ -0,0 +1,20 @@
+package contextvalue
+
+import (
+	"github.com/cschleiden/go-workflows/internal/sync"
+	"github.com/cschleiden/go-workflows/registry"
+)
+
+type registryKey struct{}
+
+func WithRegistry(ctx sync.Context, r *registry.Registry) sync.Context {
+	return sync.WithValue(ctx, registryKey{}, r)
+}
+
+func GetRegistry(ctx sync.Context) *registry.Registry {
+	if v := ctx.Value(registryKey{}); v != nil {
+		return v.(*registry.Registry)
+	}
+
+	return nil
+}

internal/contextvalue/singleworkermode.go

@@ -0,0 +1,21 @@
+package contextvalue
+
+import (
+	"github.com/cschleiden/go-workflows/internal/sync"
+)
+
+type singleWorkerModeKey struct{}
+
+// WithSingleWorkerMode sets the single worker mode flag in the context.
+// When this flag is set, activities and workflows will be automatically registered when needed.
+func WithSingleWorkerMode(ctx sync.Context) sync.Context {
+	return sync.WithValue(ctx, singleWorkerModeKey{}, true)
+}
+
+// IsSingleWorkerMode checks if the context is in single worker mode.
+func IsSingleWorkerMode(ctx sync.Context) bool {
+	if v := ctx.Value(singleWorkerModeKey{}); v != nil {
+		return v.(bool)
+	}
+	return false
+}

internal/worker/workflow.go

@@ -27,6 +27,7 @@ type WorkflowWorkerOptions struct {
 	WorkflowExecutorCache     executor.Cache
 	WorkflowExecutorCacheSize int
 	WorkflowExecutorCacheTTL  time.Duration
+	SingleWorkerMode          bool
 }
 
 func NewWorkflowWorker(
@@ -39,20 +40,22 @@ func NewWorkflowWorker(
 	}
 
 	tw := &WorkflowTaskWorker{
-		backend:  b,
-		registry: registry,
-		cache:    options.WorkflowExecutorCache,
-		logger:   b.Options().Logger,
+		backend:          b,
+		registry:         registry,
+		cache:            options.WorkflowExecutorCache,
+		logger:           b.Options().Logger,
+		singleWorkerMode: options.SingleWorkerMode,
 	}
 
 	return NewWorker(b, tw, &options.WorkerOptions)
 }
 
 type WorkflowTaskWorker struct {
-	backend  backend.Backend
-	registry *registry.Registry
-	cache    executor.Cache
-	logger   *slog.Logger
+	backend          backend.Backend
+	registry         *registry.Registry
+	cache            executor.Cache
+	logger           *slog.Logger
+	singleWorkerMode bool
 }
 
 func (wtw *WorkflowTaskWorker) Start(ctx context.Context, queues []workflow.Queue) error {
@@ -181,6 +184,7 @@ func (wtw *WorkflowTaskWorker) getExecutor(ctx context.Context, t *backend.Workf
 			t.Metadata,
 			clock.New(),
 			wtw.backend.Options().MaxHistorySize,
+			wtw.singleWorkerMode,
 		)
 		if err != nil {
 			return nil, fmt.Errorf("creating workflow task executor: %w", err)

registry/registry.go

@@ -8,29 +8,30 @@ import (
 
 	"github.com/cschleiden/go-workflows/internal/args"
 	"github.com/cschleiden/go-workflows/internal/fn"
-	wf "github.com/cschleiden/go-workflows/workflow"
 )
 
 type Registry struct {
+	sealed bool
+
 	sync.Mutex
 
-	workflowMap map[string]wf.Workflow
-	activityMap map[string]interface{}
+	workflowMap map[string]any
+	activityMap map[string]any
 }
 
 // New creates a new registry instance.
 func New() *Registry {
 	return &Registry{
-		workflowMap: make(map[string]wf.Workflow),
-		activityMap: make(map[string]interface{}),
+		workflowMap: make(map[string]any),
+		activityMap: make(map[string]any),
 	}
 }
 
 type registerConfig struct {
 	Name string
 }
 
-func (r *Registry) RegisterWorkflow(workflow wf.Workflow, opts ...RegisterOption) error {
+func (r *Registry) RegisterWorkflow(workflow any, opts ...RegisterOption) error {
 	cfg := registerOptions(opts).applyRegisterOptions(registerConfig{})
 	name := cfg.Name
 	if name == "" {
@@ -75,7 +76,7 @@ func (r *Registry) RegisterWorkflow(workflow wf.Workflow, opts ...RegisterOption
 	return nil
 }
 
-func (r *Registry) RegisterActivity(activity wf.Activity, opts ...RegisterOption) error {
+func (r *Registry) RegisterActivity(activity any, opts ...RegisterOption) error {
 	cfg := registerOptions(opts).applyRegisterOptions(registerConfig{})
 
 	t := reflect.TypeOf(activity)
@@ -151,7 +152,7 @@ func checkActivity(actType reflect.Type) error {
 	return nil
 }
 
-func (r *Registry) GetWorkflow(name string) (wf.Workflow, error) {
+func (r *Registry) GetWorkflow(name string) (any, error) {
 	r.Lock()
 	defer r.Unlock()
 

registry/registry_test.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/cschleiden/go-workflows/internal/fn"
 	"github.com/cschleiden/go-workflows/internal/sync"
-	wf "github.com/cschleiden/go-workflows/workflow"
 	"github.com/stretchr/testify/require"
 )
 
@@ -17,7 +16,7 @@ func reg_workflow1(ctx sync.Context) error {
 func TestRegistry_RegisterWorkflow(t *testing.T) {
 	type args struct {
 		name     string
-		workflow wf.Workflow
+		workflow any
 	}
 	tests := []struct {
 		name     string

samples/orchestrator/main.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"github.com/cschleiden/go-workflows/client"
+	"github.com/cschleiden/go-workflows/samples"
+	"github.com/cschleiden/go-workflows/worker"
+	"github.com/cschleiden/go-workflows/workflow"
+
+	"github.com/google/uuid"
+)
+
+func main() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	backend := samples.GetBackend("orchestrator")
+
+	orchestrator := worker.NewWorkflowOrchestrator(
+		backend,
+		nil,
+	)
+
+	if err := orchestrator.Start(ctx); err != nil {
+		panic("could not start orchestrator: " + err.Error())
+	}
+
+	// Create instance ID
+	instanceID := uuid.NewString()
+
+	// Create and run workflow using the orchestrator - no explicit registration needed
+	instance, err := orchestrator.CreateWorkflowInstance(ctx, client.WorkflowInstanceOptions{
+		InstanceID: instanceID,
+	}, SimpleWorkflow, "Hello from orchestrator!")
+	if err != nil {
+		panic("could not create workflow instance: " + err.Error())
+	}
+
+	// Wait for result using client package directly
+	result, err := client.GetWorkflowResult[string](ctx, orchestrator.Client, instance, 10*time.Second)
+	if err != nil {
+		panic("error getting workflow result: " + err.Error())
+	}
+
+	log.Printf("Workflow result: %s\n", result)
+
+	// Clean shutdown
+	cancel()
+
+	if err := orchestrator.WaitForCompletion(); err != nil {
+		panic("could not stop orchestrator: " + err.Error())
+	}
+}
+
+// SimpleWorkflow is a basic workflow that calls an activity and returns its result
+func SimpleWorkflow(ctx workflow.Context, message string) (string, error) {
+	future := workflow.ExecuteActivity[string](ctx, workflow.DefaultActivityOptions, ProcessMessage, message)
+
+	result, err := future.Get(ctx)
+	if err != nil {
+		return "", err
+	}
+
+	return result, nil
+}
+
+// ProcessMessage is a simple activity that processes a message
+// When orchestrator mode is enabled, this will be automatically registered
+func ProcessMessage(ctx context.Context, message string) (string, error) {
+	return message + " (processed by activity)", nil
+}
+
+// Note: No separate activity function needed when using InlineActivity

tester/options.go

@@ -9,12 +9,13 @@ import (
 )
 
 type options struct {
-	TestTimeout    time.Duration
-	Logger         *slog.Logger
-	Converter      converter.Converter
-	Propagators    []workflow.ContextPropagator
-	InitialTime    time.Time
-	MaxHistorySize int64
+	TestTimeout      time.Duration
+	Logger           *slog.Logger
+	Converter        converter.Converter
+	Propagators      []workflow.ContextPropagator
+	InitialTime      time.Time
+	MaxHistorySize   int64
+	SingleWorkerMode bool
 }
 
 type WorkflowTesterOption func(*options)
@@ -54,3 +55,9 @@ func WithMaxHistorySize(size int64) WorkflowTesterOption {
 		o.MaxHistorySize = size
 	}
 }
+
+func WithSingleWorkerMode(enabled bool) WorkflowTesterOption {
+	return func(o *options) {
+		o.SingleWorkerMode = enabled
+	}
+}

tester/tester.go

@@ -369,6 +369,7 @@ func (wt *workflowTester[TResult]) Execute(ctx context.Context, args ...any) {
 				tw.metadata,
 				wt.clock,
 				wt.options.MaxHistorySize,
+				wt.options.SingleWorkerMode,
 			)
 			if err != nil {
 				panic(fmt.Errorf("could not create workflow executor: %v", err))