Reduce wrapper functions, reorganize packages

Author: cschleiden

internal/tester/tester.go

@@ -32,9 +32,9 @@ type WorkflowTester interface {
 
 	Registry() *workflow.Registry
 
-	OnActivity(activity workflow.Activity, args ...interface{}) *mock.Call
+	OnActivity(activity interface{}, args ...interface{}) *mock.Call
 
-	OnSubWorkflow(workflow workflow.Workflow, args ...interface{}) *mock.Call
+	OnSubWorkflow(workflow interface{}, args ...interface{}) *mock.Call
 
 	SignalWorkflow(signalName string, value interface{})
 
@@ -78,7 +78,7 @@ type workflowTester struct {
 	options *options
 
 	// Workflow under test
-	wf  workflow.Workflow
+	wf  interface{}
 	wfi core.WorkflowInstance
 
 	// Workflows
@@ -107,7 +107,7 @@ type workflowTester struct {
 	runningActivities int32
 }
 
-func NewWorkflowTester(wf workflow.Workflow) WorkflowTester {
+func NewWorkflowTester(wf interface{}) WorkflowTester {
 	// Start with the current wall-clock tiem
 	clock := clock.NewMock()
 	clock.Set(time.Now())
@@ -164,7 +164,7 @@ func (wt *workflowTester) ListenSubWorkflow(listener func(core.WorkflowInstance,
 	wt.subWorkflowListener = listener
 }
 
-func (wt *workflowTester) OnActivity(activity workflow.Activity, args ...interface{}) *mock.Call {
+func (wt *workflowTester) OnActivity(activity interface{}, args ...interface{}) *mock.Call {
 	// Register activity so that we can correctly identify its arguments later
 	wt.registry.RegisterActivity(activity)
 
@@ -173,7 +173,7 @@ func (wt *workflowTester) OnActivity(activity workflow.Activity, args ...interfa
 	return wt.ma.On(name, args...)
 }
 
-func (wt *workflowTester) OnSubWorkflow(workflow workflow.Workflow, args ...interface{}) *mock.Call {
+func (wt *workflowTester) OnSubWorkflow(workflow interface{}, args ...interface{}) *mock.Call {
 	// Register workflow so that we can correctly identify its arguments later
 	wt.registry.RegisterWorkflow(workflow)
 
@@ -573,7 +573,7 @@ func (wt *workflowTester) scheduleSubWorkflow(event history.WorkflowEvent) {
 	}
 }
 
-func (wt *workflowTester) getInitialEvent(wf workflow.Workflow, args []interface{}) history.Event {
+func (wt *workflowTester) getInitialEvent(wf interface{}, args []interface{}) history.Event {
 	name := fn.Name(wf)
 
 	inputs, err := margs.ArgsToInputs(converter.DefaultConverter, args...)

internal/workflow/executor.go

@@ -15,6 +15,7 @@ import (
 	"github.com/cschleiden/go-workflows/internal/payload"
 	"github.com/cschleiden/go-workflows/internal/sync"
 	"github.com/cschleiden/go-workflows/internal/task"
+	"github.com/cschleiden/go-workflows/internal/workflowstate"
 	"github.com/google/uuid"
 	errs "github.com/pkg/errors"
 )
@@ -28,7 +29,7 @@ type WorkflowExecutor interface {
 type executor struct {
 	registry          *Registry
 	workflow          *workflow
-	workflowState     *workflowState
+	workflowState     *workflowstate.WfState
 	workflowCtx       sync.Context
 	workflowCtxCancel sync.CancelFunc
 	clock             clock.Clock
@@ -37,12 +38,12 @@ type executor struct {
 }
 
 func NewExecutor(registry *Registry, instance core.WorkflowInstance, clock clock.Clock) (WorkflowExecutor, error) {
-	state := newWorkflowState(instance, clock)
-	wfCtx, cancel := sync.WithCancel(WithWorkflowState(sync.Background(), state))
+	s := workflowstate.NewWorkflowState(instance, clock)
+	wfCtx, cancel := sync.WithCancel(workflowstate.WithWorkflowState(sync.Background(), s))
 
 	return &executor{
 		registry:          registry,
-		workflowState:     state,
+		workflowState:     s,
 		workflowCtx:       wfCtx,
 		workflowCtxCancel: cancel,
 		clock:             clock,
@@ -63,7 +64,7 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) ([]history
 		e.workflowState.ClearCommands()
 	} else {
 		// Replay history
-		e.workflowState.setReplaying(true)
+		e.workflowState.SetReplaying(true)
 		for _, event := range t.History {
 			if err := e.executeEvent(event); err != nil {
 				return nil, nil, errs.Wrap(err, "error while replaying event")
@@ -95,7 +96,7 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) ([]history
 }
 
 func (e *executor) executeNewEvents(newEvents []history.Event) error {
-	e.workflowState.setReplaying(false)
+	e.workflowState.SetReplaying(false)
 
 	for _, event := range newEvents {
 		if err := e.executeEvent(event); err != nil {
@@ -198,7 +199,7 @@ func (e *executor) handleWorkflowCanceled() error {
 }
 
 func (e *executor) handleWorkflowTaskStarted(event history.Event, a *history.WorkflowTaskStartedAttributes) error {
-	e.workflowState.setTime(event.Timestamp)
+	e.workflowState.SetTime(event.Timestamp)
 
 	return nil
 }
@@ -217,7 +218,7 @@ func (e *executor) handleActivityScheduled(event history.Event, a *history.Activ
 }
 
 func (e *executor) handleActivityCompleted(event history.Event, a *history.ActivityCompletedAttributes) error {
-	f, ok := e.workflowState.pendingFutures[event.ScheduleEventID]
+	f, ok := e.workflowState.FutureByScheduleEventID(event.ScheduleEventID)
 	if !ok {
 		return nil
 	}
@@ -229,7 +230,7 @@ func (e *executor) handleActivityCompleted(event history.Event, a *history.Activ
 }
 
 func (e *executor) handleActivityFailed(event history.Event, a *history.ActivityFailedAttributes) error {
-	f, ok := e.workflowState.pendingFutures[event.ScheduleEventID]
+	f, ok := e.workflowState.FutureByScheduleEventID(event.ScheduleEventID)
 	if !ok {
 		return errors.New("no pending future found for activity failed event")
 	}
@@ -248,7 +249,7 @@ func (e *executor) handleTimerScheduled(event history.Event, a *history.TimerSch
 }
 
 func (e *executor) handleTimerFired(event history.Event, a *history.TimerFiredAttributes) error {
-	f, ok := e.workflowState.pendingFutures[event.ScheduleEventID]
+	f, ok := e.workflowState.FutureByScheduleEventID(event.ScheduleEventID)
 	if !ok {
 		// Timer already canceled ignore
 		return nil
@@ -274,7 +275,7 @@ func (e *executor) handleSubWorkflowScheduled(event history.Event, a *history.Su
 }
 
 func (e *executor) handleSubWorkflowFailed(event history.Event, a *history.SubWorkflowFailedAttributes) error {
-	f, ok := e.workflowState.pendingFutures[event.ScheduleEventID]
+	f, ok := e.workflowState.FutureByScheduleEventID(event.ScheduleEventID)
 	if !ok {
 		return errors.New("no pending future found for sub workflow failed event")
 	}
@@ -287,7 +288,7 @@ func (e *executor) handleSubWorkflowFailed(event history.Event, a *history.SubWo
 }
 
 func (e *executor) handleSubWorkflowCompleted(event history.Event, a *history.SubWorkflowCompletedAttributes) error {
-	f, ok := e.workflowState.pendingFutures[event.ScheduleEventID]
+	f, ok := e.workflowState.FutureByScheduleEventID(event.ScheduleEventID)
 	if !ok {
 		return errors.New("no pending future found for sub workflow completed event")
 	}
@@ -310,7 +311,7 @@ func (e *executor) handleSignalReceived(event history.Event, a *history.SignalRe
 }
 
 func (e *executor) handleSideEffectResult(event history.Event, a *history.SideEffectResultAttributes) error {
-	f, ok := e.workflowState.pendingFutures[event.ScheduleEventID]
+	f, ok := e.workflowState.FutureByScheduleEventID(event.ScheduleEventID)
 	if !ok {
 		return errors.New("no pending future found for side effect result event")
 	}
@@ -321,8 +322,7 @@ func (e *executor) handleSideEffectResult(event history.Event, a *history.SideEf
 }
 
 func (e *executor) workflowCompleted(result payload.Payload, err error) error {
-	eventId := e.workflowState.scheduleEventID
-	e.workflowState.scheduleEventID++
+	eventId := e.workflowState.GetNextScheduleEventID()
 
 	cmd := command.NewCompleteWorkflowCommand(eventId, result, err)
 	e.workflowState.AddCommand(&cmd)
@@ -332,7 +332,7 @@ func (e *executor) workflowCompleted(result payload.Payload, err error) error {
 
 func (e *executor) processCommands(ctx context.Context, t *task.Workflow) ([]history.Event, []history.WorkflowEvent, error) {
 	instance := t.WorkflowInstance
-	commands := e.workflowState.commands
+	commands := e.workflowState.Commands()
 
 	newEvents := make([]history.Event, 0)
 	workflowEvents := make([]history.WorkflowEvent, 0)

internal/workflow/executor_test.go

@@ -16,17 +16,19 @@ import (
 	"github.com/cschleiden/go-workflows/internal/payload"
 	"github.com/cschleiden/go-workflows/internal/sync"
 	"github.com/cschleiden/go-workflows/internal/task"
+	"github.com/cschleiden/go-workflows/internal/workflowstate"
+	wf "github.com/cschleiden/go-workflows/workflow"
 	"github.com/stretchr/testify/require"
 )
 
 func newExecutor(r *Registry, i core.WorkflowInstance) *executor {
-	state := newWorkflowState(i, clock.New())
-	wfCtx, cancel := sync.WithCancel(WithWorkflowState(sync.Background(), state))
+	s := workflowstate.NewWorkflowState(i, clock.New())
+	wfCtx, cancel := sync.WithCancel(workflowstate.WithWorkflowState(sync.Background(), s))
 
 	return &executor{
 		registry:          r,
 		workflow:          NewWorkflow(reflect.ValueOf(workflow1)),
-		workflowState:     state,
+		workflowState:     s,
 		workflowCtx:       wfCtx,
 		workflowCtxCancel: cancel,
 		logger:            log.Default(),
@@ -74,15 +76,15 @@ func Test_ExecuteWorkflow(t *testing.T) {
 
 	require.Equal(t, 1, workflowHits)
 	require.True(t, e.workflow.Completed())
-	require.Len(t, e.workflowState.commands, 1)
+	require.Len(t, e.workflowState.Commands(), 1)
 }
 
 var workflowActivityHit int
 
 func workflowWithActivity(ctx sync.Context) error {
 	workflowActivityHit++
 
-	f1 := ExecuteActivity(ctx, DefaultActivityOptions, activity1, 42)
+	f1 := wf.ExecuteActivity(ctx, wf.DefaultActivityOptions, activity1, 42)
 
 	var r int
 	err := f1.Get(ctx, &r)
@@ -144,7 +146,7 @@ func Test_ReplayWorkflowWithActivityResult(t *testing.T) {
 
 	require.Equal(t, 2, workflowActivityHit)
 	require.True(t, e.workflow.Completed())
-	require.Len(t, e.workflowState.commands, 1)
+	require.Len(t, e.workflowState.Commands(), 1)
 }
 
 func Test_ExecuteWorkflowWithActivityCommand(t *testing.T) {
@@ -174,7 +176,7 @@ func Test_ExecuteWorkflowWithActivityCommand(t *testing.T) {
 	e.ExecuteTask(context.Background(), task)
 
 	require.Equal(t, 1, workflowActivityHit)
-	require.Len(t, e.workflowState.commands, 1)
+	require.Len(t, e.workflowState.Commands(), 1)
 
 	inputs, _ := converter.DefaultConverter.To(42)
 	require.Equal(t, command.Command{
@@ -185,7 +187,7 @@ func Test_ExecuteWorkflowWithActivityCommand(t *testing.T) {
 			Name:   "activity1",
 			Inputs: []payload.Payload{inputs},
 		},
-	}, *e.workflowState.commands[0])
+	}, *e.workflowState.Commands()[0])
 }
 
 var workflowTimerHits int
@@ -194,7 +196,7 @@ func workflowWithTimer(ctx sync.Context) error {
 	workflowTimerHits++
 
 	var r bool
-	if err := ScheduleTimer(ctx, time.Millisecond*5).Get(ctx, &r); err != nil {
+	if err := wf.ScheduleTimer(ctx, time.Millisecond*5).Get(ctx, &r); err != nil {
 		panic("error getting timer future")
 	}
 
@@ -229,19 +231,19 @@ func Test_ExecuteWorkflowWithTimer(t *testing.T) {
 	e.ExecuteTask(context.Background(), task)
 
 	require.Equal(t, 1, workflowTimerHits)
-	require.Len(t, e.workflowState.commands, 1)
+	require.Len(t, e.workflowState.Commands(), 1)
 
-	require.Equal(t, 1, e.workflowState.commands[0].ID)
-	require.Equal(t, command.CommandType_ScheduleTimer, e.workflowState.commands[0].Type)
+	require.Equal(t, 1, e.workflowState.Commands()[0].ID)
+	require.Equal(t, command.CommandType_ScheduleTimer, e.workflowState.Commands()[0].Type)
 }
 
 var workflowWithSelectorHits int
 
 func workflowWithSelector(ctx sync.Context) error {
 	workflowWithSelectorHits++
 
-	f1 := ExecuteActivity(ctx, DefaultActivityOptions, activity1, 42)
-	t := ScheduleTimer(ctx, time.Millisecond*2)
+	f1 := wf.ExecuteActivity(ctx, wf.DefaultActivityOptions, activity1, 42)
+	t := wf.ScheduleTimer(ctx, time.Millisecond*2)
 
 	sync.Select(
 		ctx,
@@ -284,10 +286,10 @@ func Test_ExecuteWorkflowWithSelector(t *testing.T) {
 	e.ExecuteTask(context.Background(), task)
 
 	require.Equal(t, 1, workflowWithSelectorHits)
-	require.Len(t, e.workflowState.commands, 2)
+	require.Len(t, e.workflowState.Commands(), 2)
 
-	require.Equal(t, command.CommandType_ScheduleTimer, e.workflowState.commands[0].Type)
-	require.Equal(t, command.CommandType_ScheduleActivityTask, e.workflowState.commands[1].Type)
+	require.Equal(t, command.CommandType_ScheduleTimer, e.workflowState.Commands()[0].Type)
+	require.Equal(t, command.CommandType_ScheduleActivityTask, e.workflowState.Commands()[1].Type)
 }
 
 func Test_ExecuteNewEvents(t *testing.T) {
@@ -332,7 +334,7 @@ func Test_ExecuteNewEvents(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, 1, workflowActivityHit)
 	require.False(t, e.workflow.Completed())
-	require.Len(t, e.workflowState.commands, 0)
+	require.Len(t, e.workflowState.Commands(), 0)
 
 	h := []history.Event{}
 	h = append(h, oldTask.NewEvents...)
@@ -360,14 +362,14 @@ func Test_ExecuteNewEvents(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, 2, workflowActivityHit)
 	require.True(t, e.workflow.Completed())
-	require.Len(t, e.workflowState.commands, 1)
+	require.Len(t, e.workflowState.Commands(), 1)
 }
 
 var workflowSignalHits int
 
 func workflowWithSignal1(ctx sync.Context) error {
 
-	c := NewSignalChannel(ctx, "signal1")
+	c := wf.NewSignalChannel(ctx, "signal1")
 	c.Receive(ctx, nil)
 
 	workflowSignalHits++
@@ -409,5 +411,5 @@ func Test_ExecuteWorkflowWithSignal(t *testing.T) {
 
 	require.Equal(t, 1, workflowSignalHits)
 	require.True(t, e.workflow.Completed())
-	require.Len(t, e.workflowState.commands, 1)
+	require.Len(t, e.workflowState.Commands(), 1)
 }

internal/workflow/instance.go

@@ -12,14 +12,14 @@ type Registry struct {
 	sync.Mutex
 
 	workflowMap map[string]Workflow
-	activityMap map[string]Activity
+	activityMap map[string]interface{}
 }
 
 func NewRegistry() *Registry {
 	return &Registry{
 		Mutex:       sync.Mutex{},
 		workflowMap: make(map[string]Workflow),
-		activityMap: make(map[string]Activity),
+		activityMap: make(map[string]interface{}),
 	}
 }
 
@@ -33,7 +33,7 @@ func (r *Registry) RegisterWorkflow(workflow Workflow) error {
 	return nil
 }
 
-func (r *Registry) RegisterActivity(activity Activity) error {
+func (r *Registry) RegisterActivity(activity interface{}) error {
 	r.Lock()
 	defer r.Unlock()
 
@@ -82,7 +82,7 @@ func (r *Registry) GetWorkflow(name string) (Workflow, error) {
 	return nil, errors.New("workflow not found")
 }
 
-func (r *Registry) GetActivity(name string) (Activity, error) {
+func (r *Registry) GetActivity(name string) (interface{}, error) {
 	r.Lock()
 	defer r.Unlock()