Skip to content

Commit d589943

Browse files
cschleidencschleiden
committed
Return structs
1 parent 0f51437 commit d589943

File tree

2 files changed

+21
-32
lines changed

2 files changed

+21
-32
lines changed

‎internal/worker/activity.go‎

Lines changed: 9 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -14,12 +14,7 @@ import (
1414
"github.com/cschleiden/go-workflows/internal/workflow"
1515
)
1616

17-
type ActivityWorker interface {
18-
Start(context.Context) error
19-
WaitForCompletion() error
20-
}
21-
22-
type activityWorker struct {
17+
type ActivityWorker struct {
2318
backend backend.Backend
2419

2520
options *Options
@@ -32,8 +27,8 @@ type activityWorker struct {
3227
clock clock.Clock
3328
}
3429

35-
func NewActivityWorker(backend backend.Backend, registry *workflow.Registry, clock clock.Clock, options *Options) ActivityWorker {
36-
return &activityWorker{
30+
func NewActivityWorker(backend backend.Backend, registry *workflow.Registry, clock clock.Clock, options *Options) *ActivityWorker {
31+
return &ActivityWorker{
3732
backend: backend,
3833

3934
options: options,
@@ -47,7 +42,7 @@ func NewActivityWorker(backend backend.Backend, registry *workflow.Registry, clo
4742
}
4843
}
4944

50-
func (aw *activityWorker) Start(ctx context.Context) error {
45+
func (aw *ActivityWorker) Start(ctx context.Context) error {
5146
for i := 0; i <= aw.options.ActivityPollers; i++ {
5247
go aw.runPoll(ctx)
5348
}
@@ -57,15 +52,15 @@ func (aw *activityWorker) Start(ctx context.Context) error {
5752
return nil
5853
}
5954

60-
func (aw *activityWorker) WaitForCompletion() error {
55+
func (aw *ActivityWorker) WaitForCompletion() error {
6156
close(aw.activityTaskQueue)
6257

6358
aw.wg.Wait()
6459

6560
return nil
6661
}
6762

68-
func (aw *activityWorker) runPoll(ctx context.Context) {
63+
func (aw *ActivityWorker) runPoll(ctx context.Context) {
6964
for {
7065
select {
7166
case <-ctx.Done():
@@ -84,7 +79,7 @@ func (aw *activityWorker) runPoll(ctx context.Context) {
8479
}
8580
}
8681

87-
func (aw *activityWorker) runDispatcher(ctx context.Context) {
82+
func (aw *ActivityWorker) runDispatcher(ctx context.Context) {
8883
var sem chan struct{}
8984
if aw.options.MaxParallelActivityTasks > 0 {
9085
sem = make(chan struct{}, aw.options.MaxParallelActivityTasks)
@@ -112,7 +107,7 @@ func (aw *activityWorker) runDispatcher(ctx context.Context) {
112107
}
113108
}
114109

115-
func (aw *activityWorker) handleTask(ctx context.Context, task *task.Activity) {
110+
func (aw *ActivityWorker) handleTask(ctx context.Context, task *task.Activity) {
116111
// Start heartbeat while activity is running
117112
heartbeatCtx, cancelHeartbeat := context.WithCancel(ctx)
118113
go func(ctx context.Context) {
@@ -161,7 +156,7 @@ func (aw *activityWorker) handleTask(ctx context.Context, task *task.Activity) {
161156
}
162157
}
163158

164-
func (aw *activityWorker) poll(ctx context.Context, timeout time.Duration) (*task.Activity, error) {
159+
func (aw *ActivityWorker) poll(ctx context.Context, timeout time.Duration) (*task.Activity, error) {
165160
if timeout == 0 {
166161
timeout = 30 * time.Second
167162
}

‎internal/worker/workflow.go‎

Lines changed: 12 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -15,13 +15,7 @@ import (
1515
"github.com/cschleiden/go-workflows/log"
1616
)
1717

18-
type WorkflowWorker interface {
19-
Start(context.Context) error
20-
21-
WaitForCompletion() error
22-
}
23-
24-
type workflowWorker struct {
18+
type WorkflowWorker struct {
2519
backend backend.Backend
2620

2721
options *Options
@@ -37,15 +31,15 @@ type workflowWorker struct {
3731
wg *sync.WaitGroup
3832
}
3933

40-
func NewWorkflowWorker(backend backend.Backend, registry *workflow.Registry, options *Options) WorkflowWorker {
34+
func NewWorkflowWorker(backend backend.Backend, registry *workflow.Registry, options *Options) *WorkflowWorker {
4135
var c workflow.ExecutorCache
4236
if options.WorkflowExecutorCache != nil {
4337
c = options.WorkflowExecutorCache
4438
} else {
4539
c = cache.NewWorkflowExecutorLRUCache(options.WorkflowExecutorCacheSize, options.WorkflowExecutorCacheTTL)
4640
}
4741

48-
return &workflowWorker{
42+
return &WorkflowWorker{
4943
backend: backend,
5044

5145
options: options,
@@ -61,7 +55,7 @@ func NewWorkflowWorker(backend backend.Backend, registry *workflow.Registry, opt
6155
}
6256
}
6357

64-
func (ww *workflowWorker) Start(ctx context.Context) error {
58+
func (ww *WorkflowWorker) Start(ctx context.Context) error {
6559
for i := 0; i <= ww.options.WorkflowPollers; i++ {
6660
go ww.runPoll(ctx)
6761
}
@@ -71,15 +65,15 @@ func (ww *workflowWorker) Start(ctx context.Context) error {
7165
return nil
7266
}
7367

74-
func (ww *workflowWorker) WaitForCompletion() error {
68+
func (ww *WorkflowWorker) WaitForCompletion() error {
7569
close(ww.workflowTaskQueue)
7670

7771
ww.wg.Wait()
7872

7973
return nil
8074
}
8175

82-
func (ww *workflowWorker) runPoll(ctx context.Context) {
76+
func (ww *WorkflowWorker) runPoll(ctx context.Context) {
8377
for {
8478
select {
8579
case <-ctx.Done():
@@ -99,7 +93,7 @@ func (ww *workflowWorker) runPoll(ctx context.Context) {
9993
}
10094
}
10195

102-
func (ww *workflowWorker) runDispatcher() {
96+
func (ww *WorkflowWorker) runDispatcher() {
10397
var sem chan (struct{})
10498

10599
if ww.options.MaxParallelWorkflowTasks > 0 {
@@ -128,7 +122,7 @@ func (ww *workflowWorker) runDispatcher() {
128122
}
129123
}
130124

131-
func (ww *workflowWorker) handle(ctx context.Context, t *task.Workflow) {
125+
func (ww *WorkflowWorker) handle(ctx context.Context, t *task.Workflow) {
132126
result, err := ww.handleTask(ctx, t)
133127
if err != nil {
134128
ww.logger.Panic("could not handle workflow task", "error", err)
@@ -145,7 +139,7 @@ func (ww *workflowWorker) handle(ctx context.Context, t *task.Workflow) {
145139
}
146140
}
147141

148-
func (ww *workflowWorker) handleTask(
142+
func (ww *WorkflowWorker) handleTask(
149143
ctx context.Context,
150144
t *task.Workflow,
151145
) (*workflow.ExecutionResult, error) {
@@ -169,7 +163,7 @@ func (ww *workflowWorker) handleTask(
169163
return result, nil
170164
}
171165

172-
func (ww *workflowWorker) getExecutor(ctx context.Context, t *task.Workflow) (workflow.WorkflowExecutor, error) {
166+
func (ww *WorkflowWorker) getExecutor(ctx context.Context, t *task.Workflow) (workflow.WorkflowExecutor, error) {
173167
// Try to get a cached executor
174168
executor, ok, err := ww.cache.Get(ctx, t.WorkflowInstance)
175169
if err != nil {
@@ -192,7 +186,7 @@ func (ww *workflowWorker) getExecutor(ctx context.Context, t *task.Workflow) (wo
192186
return executor, nil
193187
}
194188

195-
func (ww *workflowWorker) heartbeatTask(ctx context.Context, task *task.Workflow) {
189+
func (ww *WorkflowWorker) heartbeatTask(ctx context.Context, task *task.Workflow) {
196190
t := time.NewTicker(ww.options.WorkflowHeartbeatInterval)
197191
defer t.Stop()
198192

@@ -208,7 +202,7 @@ func (ww *workflowWorker) heartbeatTask(ctx context.Context, task *task.Workflow
208202
}
209203
}
210204

211-
func (ww *workflowWorker) poll(ctx context.Context, timeout time.Duration) (*task.Workflow, error) {
205+
func (ww *WorkflowWorker) poll(ctx context.Context, timeout time.Duration) (*task.Workflow, error) {
212206
if timeout == 0 {
213207
timeout = 30 * time.Second
214208
}

0 commit comments

Comments
 (0)