Merge pull request #245 from lovromazgon/fn-full

Register workflow and activity with name

Author: cschleiden

internal/fn/fn_test.go

@@ -7,8 +7,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-type foo struct {
-}
+type foo struct{}
 
 func (f *foo) DoSomething(ctx context.Context) error {
 	return nil
@@ -23,7 +22,7 @@ var f foo
 func bar(_ int) {
 }
 
-func Test_GetFunctionName(t *testing.T) {
+func Test_FuncName(t *testing.T) {
 	tests := []struct {
 		name string
 		i    interface{}
@@ -44,6 +43,11 @@ func Test_GetFunctionName(t *testing.T) {
 			i:    f.DoSomething,
 			want: "DoSomething",
 		},
+		{
+			name: "anonymous function",
+			i:    func() {},
+			want: "func1",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

internal/workflow/registry.go

@@ -2,6 +2,7 @@ package workflow
 
 import (
 	"errors"
+	"fmt"
 	"reflect"
 	"sync"
 
@@ -33,6 +34,14 @@ func (e *ErrInvalidWorkflow) Error() string {
 	return e.msg
 }
 
+type ErrWorkflowAlreadyRegistered struct {
+	msg string
+}
+
+func (e *ErrWorkflowAlreadyRegistered) Error() string {
+	return e.msg
+}
+
 type ErrInvalidActivity struct {
 	msg string
 }
@@ -41,7 +50,25 @@ func (e *ErrInvalidActivity) Error() string {
 	return e.msg
 }
 
-func (r *Registry) RegisterWorkflowByName(name string, workflow wf.Workflow) error {
+type ErrActivityAlreadyRegistered struct {
+	msg string
+}
+
+func (e *ErrActivityAlreadyRegistered) Error() string {
+	return e.msg
+}
+
+type RegisterConfig struct {
+	Name string
+}
+
+func (r *Registry) RegisterWorkflow(workflow wf.Workflow, opts ...RegisterOption) error {
+	cfg := registerOptions(opts).applyRegisterOptions(RegisterConfig{})
+	name := cfg.Name
+	if name == "" {
+		name = fn.Name(workflow)
+	}
+
 	wfType := reflect.TypeOf(workflow)
 	if wfType.Kind() != reflect.Func {
 		return &ErrInvalidWorkflow{"workflow is not a function"}
@@ -72,17 +99,17 @@ func (r *Registry) RegisterWorkflowByName(name string, workflow wf.Workflow) err
 	r.Lock()
 	defer r.Unlock()
 
+	if _, ok := r.workflowMap[name]; ok {
+		return &ErrWorkflowAlreadyRegistered{fmt.Sprintf("workflow with name %q already registered", name)}
+	}
 	r.workflowMap[name] = workflow
 
 	return nil
 }
 
-func (r *Registry) RegisterWorkflow(workflow wf.Workflow) error {
-	name := fn.Name(workflow)
-	return r.RegisterWorkflowByName(name, workflow)
-}
+func (r *Registry) RegisterActivity(activity wf.Activity, opts ...RegisterOption) error {
+	cfg := registerOptions(opts).applyRegisterOptions(RegisterConfig{})
 
-func (r *Registry) RegisterActivityByName(name string, activity interface{}) error {
 	t := reflect.TypeOf(activity)
 
 	// Activities on struct
@@ -91,27 +118,34 @@ func (r *Registry) RegisterActivityByName(name string, activity interface{}) err
 	}
 
 	// Activity as function
+	name := cfg.Name
+	if name == "" {
+		name = fn.Name(activity)
+	}
+
 	if err := checkActivity(reflect.TypeOf(activity)); err != nil {
 		return err
 	}
 
 	r.Lock()
 	defer r.Unlock()
 
+	if _, ok := r.activityMap[name]; ok {
+		return &ErrActivityAlreadyRegistered{fmt.Sprintf("activity with name %q already registered", name)}
+	}
 	r.activityMap[name] = activity
 
 	return nil
 }
 
-func (r *Registry) RegisterActivity(activity interface{}) error {
-	name := fn.Name(activity)
-	return r.RegisterActivityByName(name, activity)
-}
-
 func (r *Registry) registerActivitiesFromStruct(a interface{}) error {
 	// Enumerate functions defined on a
 	v := reflect.ValueOf(a)
 	t := v.Type()
+
+	r.Lock()
+	defer r.Unlock()
+
 	for i := 0; i < v.NumMethod(); i++ {
 		mv := v.Method(i)
 		mt := t.Method(i)

internal/workflow/registry_option.go

@@ -0,0 +1,27 @@
+package workflow
+
+type RegisterOption interface {
+	applyRegisterOption(RegisterConfig) RegisterConfig
+}
+
+type registerOptions []RegisterOption
+
+func (opts registerOptions) applyRegisterOptions(cfg RegisterConfig) RegisterConfig {
+	for _, opt := range opts {
+		cfg = opt.applyRegisterOption(cfg)
+	}
+	return cfg
+}
+
+type registerOptionFunc func(RegisterConfig) RegisterConfig
+
+func (f registerOptionFunc) applyRegisterOption(cfg RegisterConfig) RegisterConfig {
+	return f(cfg)
+}
+
+func WithName(name string) RegisterOption {
+	return registerOptionFunc(func(cfg RegisterConfig) RegisterConfig {
+		cfg.Name = name
+		return cfg
+	})
+}

internal/workflow/registry_test.go

@@ -84,13 +84,8 @@ func TestRegistry_RegisterWorkflow(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			r := NewRegistry()
-			var err error
 
-			if tt.args.name != "" {
-				err = r.RegisterWorkflowByName(tt.args.name, tt.args.workflow)
-			} else {
-				err = r.RegisterWorkflow(tt.args.workflow)
-			}
+			err := r.RegisterWorkflow(tt.args.workflow, WithName(tt.args.name))
 
 			if (err != nil) != tt.wantErr {
 				t.Errorf("Registry.RegisterWorkflow() error = %v, wantErr %v", err, tt.wantErr)
@@ -106,6 +101,25 @@ func TestRegistry_RegisterWorkflow(t *testing.T) {
 	}
 }
 
+func Test_RegisterWorkflow_Conflict(t *testing.T) {
+	r := NewRegistry()
+	require.NotNil(t, r)
+
+	var wantErr *ErrWorkflowAlreadyRegistered
+
+	err := r.RegisterWorkflow(reg_workflow1)
+	require.NoError(t, err)
+
+	err = r.RegisterWorkflow(reg_workflow1)
+	require.ErrorAs(t, err, &wantErr)
+
+	err = r.RegisterWorkflow(reg_workflow1, WithName("CustomName"))
+	require.NoError(t, err)
+
+	err = r.RegisterWorkflow(reg_workflow1, WithName("CustomName"))
+	require.ErrorAs(t, err, &wantErr)
+}
+
 func reg_activity(ctx context.Context) error {
 	return nil
 }
@@ -127,7 +141,7 @@ func Test_ActivityRegistration(t *testing.T) {
 	err = fn(context.Background())
 	require.NoError(t, err)
 
-	err = r.RegisterActivityByName("CustomName", reg_activity)
+	err = r.RegisterActivity(reg_activity, WithName("CustomName"))
 	require.NoError(t, err)
 
 	x, err = r.GetActivity("CustomName")
@@ -185,6 +199,25 @@ func Test_ActivityRegistrationOnStruct(t *testing.T) {
 	require.Equal(t, "test", v)
 }
 
+func Test_RegisterActivity_Conflict(t *testing.T) {
+	r := NewRegistry()
+	require.NotNil(t, r)
+
+	var wantErr *ErrActivityAlreadyRegistered
+
+	err := r.RegisterActivity(reg_activity)
+	require.NoError(t, err)
+
+	err = r.RegisterActivity(reg_activity)
+	require.ErrorAs(t, err, &wantErr)
+
+	err = r.RegisterActivity(reg_activity, WithName("CustomName"))
+	require.NoError(t, err)
+
+	err = r.RegisterActivity(reg_activity, WithName("CustomName"))
+	require.ErrorAs(t, err, &wantErr)
+}
+
 type reg_invalid_activities struct {
 	SomeValue string
 }

tester/tester.go

@@ -252,7 +252,7 @@ func (wt *workflowTester[TResult]) ListenSubWorkflow(listener func(*core.Workflo
 
 func (wt *workflowTester[TResult]) OnActivityByName(name string, activity workflow.Activity, args ...any) *mock.Call {
 	// Register activity so that we can correctly identify its arguments later
-	wt.registry.RegisterActivityByName(name, activity)
+	wt.registry.RegisterActivity(activity, wf.WithName(name))
 
 	wt.mockedActivities[name] = true
 	return wt.ma.On(name, args...)
@@ -269,7 +269,7 @@ func (wt *workflowTester[TResult]) OnActivity(activity workflow.Activity, args .
 
 func (wt *workflowTester[TResult]) OnSubWorkflowByName(name string, workflow workflow.Workflow, args ...any) *mock.Call {
 	// Register workflow so that we can correctly identify its arguments later
-	wt.registry.RegisterWorkflowByName(name, workflow)
+	wt.registry.RegisterWorkflow(workflow, wf.WithName(name))
 
 	wt.mockedWorkflows[name] = true
 	return wt.mw.On(name, args...)

worker/option.go

@@ -0,0 +1,19 @@
+package worker
+
+import workflowinternal "github.com/cschleiden/go-workflows/internal/workflow"
+
+type RegisterOption workflowinternal.RegisterOption
+
+type registerOptions []RegisterOption
+
+func (opts registerOptions) asInternalOptions() []workflowinternal.RegisterOption {
+	repacked := make([]workflowinternal.RegisterOption, len(opts))
+	for i, opt := range opts {
+		repacked[i] = workflowinternal.RegisterOption(opt)
+	}
+	return repacked
+}
+
+func WithName(name string) RegisterOption {
+	return workflowinternal.WithName(name)
+}

worker/worker.go

@@ -92,10 +92,10 @@ func (w *Worker) WaitForCompletion() error {
 	return nil
 }
 
-func (w *Worker) RegisterWorkflow(wf workflow.Workflow) error {
-	return w.registry.RegisterWorkflow(wf)
+func (w *Worker) RegisterWorkflow(wf workflow.Workflow, opts ...RegisterOption) error {
+	return w.registry.RegisterWorkflow(wf, registerOptions(opts).asInternalOptions()...)
 }
 
-func (w *Worker) RegisterActivity(a workflow.Activity) error {
-	return w.registry.RegisterActivity(a)
+func (w *Worker) RegisterActivity(a workflow.Activity, opts ...RegisterOption) error {
+	return w.registry.RegisterActivity(a, registerOptions(opts).asInternalOptions()...)
 }