Merge pull request #294 from cschleiden/workflow-activity-by-name

Execute workflows by (custom) name

Author: cschleiden

.vscode/launch.json

@@ -3,14 +3,25 @@
   // Hover to view descriptions of existing attributes.
   // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
   "version": "0.2.0",
+  "inputs": [
+    {
+      "id": "backend",
+      "type": "pickString",
+      "description": "Choose a backend",
+      "default": "memory",
+      "options": ["mysql", "sqlite", "redis", "memory"]
+    }
+  ],
   "configurations": [
-
     {
       "name": "Current sample",
       "type": "go",
       "request": "launch",
       "mode": "debug",
-      "program": "${fileDirname}"
+      "program": "${fileDirname}",
+      "args": [
+        "-backend", "${input:backend}",
+      ]
     },
     {
       "name": "Launch simple sample",

client/client.go

@@ -44,9 +44,17 @@ func New(backend backend.Backend) *Client {
 
 // CreateWorkflowInstance creates a new workflow instance of the given workflow.
 func (c *Client) CreateWorkflowInstance(ctx context.Context, options WorkflowInstanceOptions, wf workflow.Workflow, args ...any) (*workflow.Instance, error) {
-	// Check arguments
-	if err := a.ParamsMatch(wf, args...); err != nil {
-		return nil, err
+	var workflowName string
+
+	if name, ok := wf.(string); ok {
+		workflowName = name
+	} else {
+		workflowName = fn.Name(wf)
+
+		// Check arguments if actual workflow function given here
+		if err := a.ParamsMatch(wf, args...); err != nil {
+			return nil, err
+		}
 	}
 
 	inputs, err := a.ArgsToInputs(c.backend.Converter(), args...)
@@ -57,8 +65,6 @@ func (c *Client) CreateWorkflowInstance(ctx context.Context, options WorkflowIns
 	wfi := core.NewWorkflowInstance(options.InstanceID, uuid.NewString())
 	metadata := &workflow.Metadata{}
 
-	workflowName := fn.Name(wf)
-
 	// Start new span for the workflow instance
 	ctx, span := c.backend.Tracer().Start(ctx, fmt.Sprintf("CreateWorkflowInstance: %s", workflowName), trace.WithAttributes(
 		attribute.String(log.InstanceIDKey, wfi.InstanceID),

client/client_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/cschleiden/go-workflows/backend/converter"
 	"github.com/cschleiden/go-workflows/backend/history"
 	"github.com/cschleiden/go-workflows/core"
+	"github.com/cschleiden/go-workflows/internal/metrics"
 	"github.com/cschleiden/go-workflows/workflow"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/mock"
@@ -40,6 +41,38 @@ func Test_Client_CreateWorkflowInstance_ParamMismatch(t *testing.T) {
 	b.AssertExpectations(t)
 }
 
+func Test_Client_CreateWorkflowInstance_NameGiven(t *testing.T) {
+	ctx := context.Background()
+
+	b := &backend.MockBackend{}
+	b.On("Converter").Return(converter.DefaultConverter)
+	b.On("Logger").Return(slog.Default())
+	b.On("Tracer").Return(trace.NewNoopTracerProvider().Tracer("test"))
+	b.On("Metrics").Return(metrics.NewNoopMetricsClient())
+	b.On("ContextPropagators").Return(nil)
+	b.On("CreateWorkflowInstance", mock.Anything, mock.Anything, mock.MatchedBy(func(event *history.Event) bool {
+		if event.Type != history.EventType_WorkflowExecutionStarted {
+			return false
+		}
+
+		a := event.Attributes.(*history.ExecutionStartedAttributes)
+
+		return a.Name == "workflowName"
+	})).Return(nil, nil)
+
+	c := &Client{
+		backend: b,
+		clock:   clock.New(),
+	}
+
+	result, err := c.CreateWorkflowInstance(ctx, WorkflowInstanceOptions{
+		InstanceID: "id",
+	}, "workflowName", "foo")
+	require.NotZero(t, result)
+	require.NoError(t, err)
+	b.AssertExpectations(t)
+}
+
 func Test_Client_GetWorkflowResultTimeout(t *testing.T) {
 	instance := core.NewWorkflowInstance(uuid.NewString(), "test")
 

samples/workflow-registration/workflow.go

@@ -0,0 +1,34 @@
+package main
+
+import (
+	"context"
+
+	"github.com/cschleiden/go-workflows/activity"
+	"github.com/cschleiden/go-workflows/workflow"
+)
+
+func Workflow1(ctx workflow.Context, msg string) (int, error) {
+	logger := workflow.Logger(ctx)
+	logger.Debug("Entering Workflow1", "msg", msg)
+	defer logger.Debug("Leaving Workflow1")
+
+	r1, err := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity1, 35, 12).Get(ctx)
+	if err != nil {
+		panic("error getting activity 1 result")
+	}
+	logger.Debug("R1 result", "r1", r1)
+
+	return r1, nil
+}
+
+func SubWorkflow1(ctx workflow.Context, msg string) (int, error) {
+	return 42, nil
+}
+
+func Activity1(ctx context.Context, a, b int) (int, error) {
+	logger := activity.Logger(ctx)
+	logger.Debug("Entering Activity1")
+	defer logger.Debug("Leaving Activity1")
+
+	return a + b, nil
+}

samples/workflow-registration/workflow_registration.go

@@ -0,0 +1,67 @@
+package main
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"github.com/cschleiden/go-workflows/backend"
+	"github.com/cschleiden/go-workflows/client"
+	"github.com/cschleiden/go-workflows/internal/workflow"
+	"github.com/cschleiden/go-workflows/samples"
+	"github.com/cschleiden/go-workflows/worker"
+
+	"github.com/google/uuid"
+)
+
+func main() {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	b := samples.GetBackend("simple")
+
+	// Run worker
+	w := RunWorker(ctx, b)
+
+	// Start workflow via client
+	c := client.New(b)
+
+	runWorkflow(ctx, c)
+
+	cancel()
+
+	if err := w.WaitForCompletion(); err != nil {
+		panic("could not stop worker" + err.Error())
+	}
+}
+
+func runWorkflow(ctx context.Context, c *client.Client) {
+	wf, err := c.CreateWorkflowInstance(ctx, client.WorkflowInstanceOptions{
+		InstanceID: uuid.NewString(),
+	}, "WorkflowName", "Hello world")
+	if err != nil {
+		log.Fatal(err)
+		panic("could not start workflow")
+	}
+
+	result, err := client.GetWorkflowResult[int](ctx, c, wf, time.Second*10)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	log.Println("Workflow finished. Result:", result)
+}
+
+func RunWorker(ctx context.Context, mb backend.Backend) *worker.Worker {
+	w := worker.New(mb, nil)
+
+	w.RegisterWorkflow(Workflow1, workflow.WithName("WorkflowName"))
+	w.RegisterWorkflow(SubWorkflow1, workflow.WithName("SubWorkflowName"))
+
+	w.RegisterActivity(Activity1)
+
+	if err := w.Start(ctx); err != nil {
+		panic("could not start worker")
+	}
+
+	return w
+}

samples/workflow-registration/workflow_registration_test.go

@@ -0,0 +1,27 @@
+package main
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cschleiden/go-workflows/tester"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_Workflow(t *testing.T) {
+	tester := tester.NewWorkflowTester[int](Workflow1)
+
+	tester.OnSubWorkflowByName("SubWorkflowName", mock.Anything, "Hello world").Return(42, nil)
+
+	tester.OnActivity(Activity1, mock.Anything, 35, 12).Return(47, nil)
+
+	tester.Execute(context.Background(), "Hello world")
+
+	require.True(t, tester.WorkflowFinished())
+
+	wr, werr := tester.WorkflowResult()
+	require.Equal(t, 47, wr)
+	require.Empty(t, werr)
+	tester.AssertExpectations(t)
+}

workflow/subworkflow.go

@@ -50,19 +50,24 @@ func createSubWorkflowInstance[TResult any](ctx Context, options SubWorkflowOpti
 	}
 
 	// Check return type
-	if err := a.ReturnTypeMatch[TResult](wf); err != nil {
-		f.Set(*new(TResult), err)
-		return f
-	}
+	var workflowName string
+	if name, ok := wf.(string); ok {
+		workflowName = name
+	} else {
+		workflowName = fn.Name(wf)
+
+		if err := a.ReturnTypeMatch[TResult](wf); err != nil {
+			f.Set(*new(TResult), err)
+			return f
+		}
 
-	// Check arguments
-	if err := a.ParamsMatch(wf, args...); err != nil {
-		f.Set(*new(TResult), err)
-		return f
+		// Check arguments
+		if err := a.ParamsMatch(wf, args...); err != nil {
+			f.Set(*new(TResult), err)
+			return f
+		}
 	}
 
-	name := fn.Name(wf)
-
 	cv := contextvalue.Converter(ctx)
 	inputs, err := a.ArgsToInputs(cv, args...)
 	if err != nil {
@@ -74,9 +79,9 @@ func createSubWorkflowInstance[TResult any](ctx Context, options SubWorkflowOpti
 	scheduleEventID := wfState.GetNextScheduleEventID()
 
 	ctx, span := workflowtracer.Tracer(ctx).Start(ctx,
-		fmt.Sprintf("CreateSubworkflowInstance: %s", name),
+		fmt.Sprintf("CreateSubworkflowInstance: %s", workflowName),
 		trace.WithAttributes(
-			attribute.String(log.WorkflowNameKey, name),
+			attribute.String(log.WorkflowNameKey, workflowName),
 			attribute.Int64(log.ScheduleEventIDKey, scheduleEventID),
 			attribute.Int(log.AttemptKey, attempt),
 		))
@@ -90,7 +95,7 @@ func createSubWorkflowInstance[TResult any](ctx Context, options SubWorkflowOpti
 		return f
 	}
 
-	cmd := command.NewScheduleSubWorkflowCommand(scheduleEventID, wfState.Instance(), options.InstanceID, name, inputs, metadata)
+	cmd := command.NewScheduleSubWorkflowCommand(scheduleEventID, wfState.Instance(), options.InstanceID, workflowName, inputs, metadata)
 
 	wfState.AddCommand(cmd)
 	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(cv, f))

workflow/subworkflow_test.go

@@ -15,6 +15,25 @@ import (
 	"go.opentelemetry.io/otel/trace"
 )
 
+func Test_createSubWorkflowInstance_NameAsString(t *testing.T) {
+	ctx := sync.Background()
+	ctx = contextvalue.WithConverter(ctx, converter.DefaultConverter)
+	ctx = workflowstate.WithWorkflowState(
+		ctx,
+		workflowstate.NewWorkflowState(core.NewWorkflowInstance("a", ""), slog.Default(), clock.New()),
+	)
+	ctx = workflowtracer.WithWorkflowTracer(ctx, workflowtracer.New(trace.NewNoopTracerProvider().Tracer("test")))
+
+	c := sync.NewCoroutine(ctx, func(ctx Context) error {
+		createSubWorkflowInstance[int](ctx, DefaultSubWorkflowOptions, 1, "workflowName", "foo")
+		return nil
+	})
+
+	c.Execute()
+	require.NoError(t, c.Error())
+	require.True(t, c.Finished())
+}
+
 func Test_createSubWorkflowInstance_ParamMismatch(t *testing.T) {
 	wf := func(Context, int) (int, error) {
 		return 42, nil