Add single worker mode

Author: cschleiden

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

registry/registry.go

@@ -8,29 +8,28 @@ 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 {
 	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 +74,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 +150,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

worker/workflow_orchestrator.go

@@ -0,0 +1,95 @@
+package worker
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/cschleiden/go-workflows/backend"
+	"github.com/cschleiden/go-workflows/client"
+	"github.com/cschleiden/go-workflows/internal/fn"
+	"github.com/cschleiden/go-workflows/registry"
+	"github.com/cschleiden/go-workflows/workflow"
+)
+
+// WorkflowOrchestrator combines a worker and client into a single entity.
+// It orchestrates the entire workflow lifecycle, from creation to execution.
+type WorkflowOrchestrator struct {
+	worker   *Worker
+	Client   *client.Client // Exposed for direct access to GetWorkflowResult
+	registry *registry.Registry
+}
+
+// NewWorkflowOrchestrator creates a new orchestrator with client capabilities and optional registration.
+func NewWorkflowOrchestrator(backend backend.Backend, options *Options) *WorkflowOrchestrator {
+	if options == nil {
+		options = &DefaultOptions
+	}
+
+	// Create registry that will be shared between worker and orchestrator
+	reg := registry.New()
+
+	// Create a regular worker with the registry
+	workflowWorker := newWorkflowWorker(backend, reg, &options.WorkflowWorkerOptions)
+	activityWorker := newActivityWorker(backend, reg, &options.ActivityWorkerOptions)
+	w := newWorker(backend, reg, []worker{workflowWorker, activityWorker})
+	c := client.New(backend)
+
+	// Create orchestrator that combines both
+	orchestrator := &WorkflowOrchestrator{
+		worker:   w,
+		Client:   c,
+		registry: reg,
+	}
+
+	// No automatic registration - will be done on-demand
+
+	return orchestrator
+}
+
+// Start starts the worker.
+func (o *WorkflowOrchestrator) Start(ctx context.Context) error {
+	return o.worker.Start(ctx)
+}
+
+// WaitForCompletion waits for the worker to complete processing.
+func (o *WorkflowOrchestrator) WaitForCompletion() error {
+	return o.worker.WaitForCompletion()
+}
+
+// CreateWorkflowInstance creates a new workflow instance using the client.
+// Automatically registers the workflow if it's not already registered.
+func (o *WorkflowOrchestrator) CreateWorkflowInstance(ctx context.Context, options client.WorkflowInstanceOptions, wf workflow.Workflow, args ...any) (*workflow.Instance, error) {
+	// Check if the workflow is a function (not a string name) and register it if needed
+	if _, ok := wf.(string); !ok {
+		// It's a function reference, try to register it if not already registered
+		name := fn.Name(wf)
+		_, err := o.registry.GetWorkflow(name)
+		if err != nil {
+			// Workflow not found in registry, register it directly
+			if err := o.worker.RegisterWorkflow(wf); err != nil {
+				return nil, fmt.Errorf("auto-registering workflow %s: %w", name, err)
+			}
+		}
+	}
+
+	return o.Client.CreateWorkflowInstance(ctx, options, wf, args...)
+}
+
+// WaitForWorkflowInstance waits for a workflow instance to complete.
+func (o *WorkflowOrchestrator) WaitForWorkflowInstance(ctx context.Context, instance *workflow.Instance, timeout time.Duration) error {
+	return o.Client.WaitForWorkflowInstance(ctx, instance, timeout)
+}
+
+// Note: Use client.GetWorkflowResult directly with the embedded client:
+// result, err := client.GetWorkflowResult[ResultType](ctx, orchestrator.client, instance, timeout)
+
+// SignalWorkflow signals a workflow instance.
+func (o *WorkflowOrchestrator) SignalWorkflow(ctx context.Context, instanceID string, name string, arg any) error {
+	return o.Client.SignalWorkflow(ctx, instanceID, name, arg)
+}
+
+// RemoveWorkflowInstance removes a workflow instance.
+func (o *WorkflowOrchestrator) RemoveWorkflowInstance(ctx context.Context, instance *workflow.Instance) error {
+	return o.Client.RemoveWorkflowInstance(ctx, instance)
+}

workflow/executor/executor.go

@@ -81,7 +81,7 @@ type executor struct {
 func NewExecutor(
 	logger *slog.Logger,
 	tracer trace.Tracer,
-	registry *registry.Registry,
+	r *registry.Registry,
 	cv converter.Converter,
 	propagators []wf.ContextPropagator,
 	historyProvider WorkflowHistoryProvider,
@@ -96,6 +96,7 @@ func NewExecutor(
 	wfCtx = contextvalue.WithConverter(wfCtx, cv)
 	wfCtx = workflowstate.WithWorkflowState(wfCtx, s)
 	wfCtx = sync.WithValue(wfCtx, contextvalue.PropagatorsCtxKey, propagators)
+	wfCtx = contextvalue.WithRegistry(wfCtx, r)
 	wfCtx, cancel := sync.WithCancel(wfCtx)
 
 	// As part of this, the default tracing propagator will run, and set the parent span
@@ -114,7 +115,7 @@ func NewExecutor(
 	)
 
 	return &executor{
-		registry:          registry,
+		registry:          r,
 		historyProvider:   historyProvider,
 		workflowState:     s,
 		workflowCtx:       wfCtx,

workflow/executor/executor_test.go

@@ -278,10 +278,10 @@ func Test_Executor(t *testing.T) {
 
 					sync.Select(
 						ctx,
-						sync.Await[int](f1, func(ctx sync.Context, f sync.Future[int]) {
+						sync.Await(f1, func(ctx sync.Context, f sync.Future[int]) {
 							workflowWithSelectorHits++
 						}),
-						sync.Await[any](t, func(ctx sync.Context, _ sync.Future[any]) {
+						sync.Await(t, func(ctx sync.Context, _ sync.Future[any]) {
 							workflowWithSelectorHits++
 						}),
 					)

workflow/select.go

@@ -11,20 +11,20 @@ func Select(ctx Context, cases ...SelectCase) {
 
 // Await calls the provided handler when the given future is ready.
 func Await[T any](f Future[T], handler func(Context, Future[T])) SelectCase {
-	return sync.Await[T](f, func(ctx sync.Context, f sync.Future[T]) {
+	return sync.Await(f, func(ctx sync.Context, f sync.Future[T]) {
 		handler(ctx, f)
 	})
 }
 
 // Receive calls the provided handler if the given channel can receive a value. The handler receives
 // the received value, and the ok flag indicating whether the value was received or the channel was closed.
 func Receive[T any](c Channel[T], handler func(ctx Context, v T, ok bool)) SelectCase {
-	return sync.Receive[T](c, handler)
+	return sync.Receive(c, handler)
 }
 
 // Send calls the provided handler if the given value can be sent to the channel.
 func Send[T any](c Channel[T], value *T, handler func(ctx Context)) SelectCase {
-	return sync.Send[T](c, value, handler)
+	return sync.Send(c, value, handler)
 }
 
 // Default calls the provided handler if none of the other cases match.