Merge pull request #67 from cschleiden/registry-error-checks

Better error checking when registering workflows and activities

Author: cschleiden

internal/args/args.go

@@ -37,7 +37,7 @@ func InputsToArgs(c converter.Converter, fn reflect.Value, inputs []payload.Payl
 		argT := activityFnT.In(i)
 
 		// Insert context if requested
-		if i == 0 && (isOwnContext(argT) || isContext(argT)) {
+		if i == 0 && (IsOwnContext(argT) || isContext(argT)) {
 			addContext = true
 			continue
 		}
@@ -66,7 +66,7 @@ func InputsToArgs(c converter.Converter, fn reflect.Value, inputs []payload.Payl
 	return args, addContext, nil
 }
 
-func isOwnContext(inType reflect.Type) bool {
+func IsOwnContext(inType reflect.Type) bool {
 	contextElem := reflect.TypeOf((*sync.Context)(nil)).Elem()
 	return inType != nil && inType.Implements(contextElem)
 }

internal/workflow/registry.go

@@ -5,6 +5,7 @@ import (
 	"reflect"
 	"sync"
 
+	"github.com/cschleiden/go-workflows/internal/args"
 	"github.com/cschleiden/go-workflows/internal/fn"
 )
 
@@ -25,10 +26,53 @@ func NewRegistry() *Registry {
 	}
 }
 
+type ErrInvalidWorkflow struct {
+	msg string
+}
+
+func (e *ErrInvalidWorkflow) Error() string {
+	return e.msg
+}
+
+type ErrInvalidActivity struct {
+	msg string
+}
+
+func (e *ErrInvalidActivity) Error() string {
+	return e.msg
+}
+
 func (r *Registry) RegisterWorkflow(workflow Workflow) error {
 	r.Lock()
 	defer r.Unlock()
 
+	wfType := reflect.TypeOf(workflow)
+	if wfType.Kind() != reflect.Func {
+		return &ErrInvalidWorkflow{"workflow is not a function"}
+	}
+
+	if wfType.NumIn() == 0 {
+		return &ErrInvalidWorkflow{"workflow does not accept context parameter"}
+	}
+
+	if !args.IsOwnContext(wfType.In(0)) {
+		return &ErrInvalidWorkflow{"workflow does not accept context as first parameter"}
+	}
+
+	if wfType.NumOut() == 0 {
+		return &ErrInvalidWorkflow{"workflow must return error"}
+	}
+
+	if wfType.NumOut() > 2 {
+		return &ErrInvalidWorkflow{"workflow must return at most two values"}
+	}
+
+	errType := reflect.TypeOf((*error)(nil)).Elem()
+	if (wfType.NumOut() == 1 && !wfType.Out(0).Implements(errType)) ||
+		(wfType.NumOut() == 2 && !wfType.Out(1).Implements(errType)) {
+		return &ErrInvalidWorkflow{"workflow must return error as last return value"}
+	}
+
 	name := fn.Name(workflow)
 	r.workflowMap[name] = workflow
 
@@ -47,6 +91,10 @@ func (r *Registry) RegisterActivity(activity interface{}) error {
 	}
 
 	// Activity as function
+	if err := checkActivity(reflect.TypeOf(activity)); err != nil {
+		return err
+	}
+
 	name := fn.Name(activity)
 	r.activityMap[name] = activity
 
@@ -66,13 +114,34 @@ func (r *Registry) registerActivitiesFromStruct(a interface{}) error {
 			continue
 		}
 
+		if err := checkActivity(mt.Type); err != nil {
+			return err
+		}
+
 		name := mt.Name
 		r.activityMap[name] = mv.Interface()
 	}
 
 	return nil
 }
 
+func checkActivity(actType reflect.Type) error {
+	if actType.Kind() != reflect.Func {
+		return &ErrInvalidActivity{"activity not a func"}
+	}
+
+	if actType.NumOut() == 0 {
+		return &ErrInvalidActivity{"activity must return error"}
+	}
+
+	errType := reflect.TypeOf((*error)(nil)).Elem()
+	if !actType.Out(actType.NumOut() - 1).Implements(errType) {
+		return &ErrInvalidWorkflow{"activity must return error as last return value"}
+	}
+
+	return nil
+}
+
 func (r *Registry) GetWorkflow(name string) (Workflow, error) {
 	r.Lock()
 	defer r.Unlock()

internal/workflow/registry_test.go

@@ -13,22 +13,79 @@ func reg_workflow1(ctx sync.Context) error {
 	return nil
 }
 
-func Test_WorkflowRegistration(t *testing.T) {
-	r := NewRegistry()
-	require.NotNil(t, r)
-
-	err := r.RegisterWorkflow(reg_workflow1)
-	require.NoError(t, err)
-
-	x, err := r.GetWorkflow("reg_workflow1")
-	require.NoError(t, err)
-
-	fn, ok := x.(func(context sync.Context) error)
-	require.True(t, ok)
-	require.NotNil(t, fn)
-
-	err = fn(sync.Background())
-	require.NoError(t, err)
+func TestRegistry_RegisterWorkflow(t *testing.T) {
+	type args struct {
+		workflow Workflow
+	}
+	tests := []struct {
+		name     string
+		args     args
+		wantName string
+		wantErr  bool
+	}{
+		{
+			name: "valid workflow",
+			args: args{
+				workflow: reg_workflow1,
+			},
+			wantName: "reg_workflow1",
+		},
+		{
+			name: "valid workflow with results",
+			args: args{
+				workflow: func(ctx sync.Context) (int, error) { return 42, nil },
+			},
+		},
+		{
+			name: "valid workflow with multiple parameters",
+			args: args{
+				workflow: func(ctx sync.Context, a, b int) (int, error) { return 42, nil },
+			},
+		},
+		{
+			name: "missing parameter",
+			args: args{
+				workflow: func(ctx context.Context) {},
+			},
+			wantErr: true,
+		},
+		{
+			name: "missing error result",
+			args: args{
+				workflow: func(ctx sync.Context) {},
+			},
+			wantErr: true,
+		},
+		{
+			name: "missing error with results",
+			args: args{
+				workflow: func(ctx sync.Context) int { return 42 },
+			},
+			wantErr: true,
+		},
+		{
+			name: "missing error with results",
+			args: args{
+				workflow: func(ctx sync.Context) int { return 42 },
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := NewRegistry()
+			if err := r.RegisterWorkflow(tt.args.workflow); (err != nil) != tt.wantErr {
+				t.Errorf("Registry.RegisterWorkflow() error = %v, wantErr %v", err, tt.wantErr)
+				t.FailNow()
+			}
+
+			if tt.wantName != "" {
+				x, err := r.GetWorkflow(tt.wantName)
+				require.NoError(t, err)
+				require.NotNil(t, x)
+			}
+		})
+	}
 }
 
 func reg_activity(ctx context.Context) error {
@@ -53,6 +110,17 @@ func Test_ActivityRegistration(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func reg_activity_invalid(ctx context.Context) {
+}
+
+func Test_ActivityRegistration_Invalid(t *testing.T) {
+	r := NewRegistry()
+	require.NotNil(t, r)
+
+	err := r.RegisterActivity(reg_activity_invalid)
+	require.Error(t, err)
+}
+
 type reg_activities struct {
 	SomeValue string
 }
@@ -92,3 +160,21 @@ func Test_ActivityRegistrationOnStruct(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, "test", v)
 }
+
+type reg_invalid_activities struct {
+	SomeValue string
+}
+
+func (r *reg_invalid_activities) Activity1(ctx context.Context) {
+}
+
+func Test_ActivityRegistrationOnStruct_Invalid(t *testing.T) {
+	r := NewRegistry()
+	require.NotNil(t, r)
+
+	a := &reg_invalid_activities{
+		SomeValue: "test",
+	}
+	err := r.RegisterActivity(a)
+	require.Error(t, err)
+}