1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
|
package workflow
import (
"context"
"sync"
"sync/atomic"
"github.com/spiral/errors"
"github.com/spiral/roadrunner/v2/pkg/events"
"github.com/spiral/roadrunner/v2/pkg/payload"
rrWorker "github.com/spiral/roadrunner/v2/pkg/worker"
"github.com/spiral/roadrunner/v2/plugins/server"
"github.com/spiral/roadrunner/v2/plugins/temporal/client"
rrt "github.com/spiral/roadrunner/v2/plugins/temporal/protocol"
bindings "go.temporal.io/sdk/internalbindings"
"go.temporal.io/sdk/worker"
"go.temporal.io/sdk/workflow"
)
const eventWorkerExit = 8390
// RR_MODE env variable key
const RR_MODE = "RR_MODE" //nolint
// RR_CODEC env variable key
const RR_CODEC = "RR_CODEC" //nolint
type workflowPool interface {
SeqID() uint64
Exec(p payload.Payload) (payload.Payload, error)
Start(ctx context.Context, temporal client.Temporal) error
Destroy(ctx context.Context) error
Workers() []rrWorker.BaseProcess
WorkflowNames() []string
}
// PoolEvent triggered on workflow pool worker events.
type PoolEvent struct {
Event int
Context interface{}
Caused error
}
// workflowPoolImpl manages workflowProcess executions between worker restarts.
type workflowPoolImpl struct {
codec rrt.Codec
seqID uint64
workflows map[string]rrt.WorkflowInfo
tWorkers []worker.Worker
mu sync.Mutex
worker rrWorker.SyncWorker
active bool
}
// newWorkflowPool creates new workflow pool.
func newWorkflowPool(codec rrt.Codec, listener events.Listener, factory server.Server) (workflowPool, error) {
const op = errors.Op("new_workflow_pool")
w, err := factory.NewWorker(
context.Background(),
map[string]string{RR_MODE: RRMode, RR_CODEC: codec.GetName()},
listener,
)
if err != nil {
return nil, errors.E(op, err)
}
go func() {
err := w.Wait()
listener(PoolEvent{Event: eventWorkerExit, Caused: err})
}()
return &workflowPoolImpl{codec: codec, worker: rrWorker.From(w)}, nil
}
// Start the pool in non blocking mode.
func (pool *workflowPoolImpl) Start(ctx context.Context, temporal client.Temporal) error {
const op = errors.Op("workflow_pool_start")
pool.mu.Lock()
pool.active = true
pool.mu.Unlock()
err := pool.initWorkers(ctx, temporal)
if err != nil {
return errors.E(op, err)
}
for i := 0; i < len(pool.tWorkers); i++ {
err := pool.tWorkers[i].Start()
if err != nil {
return errors.E(op, err)
}
}
return nil
}
// Active.
func (pool *workflowPoolImpl) Active() bool {
return pool.active
}
// Destroy stops all temporal workers and application worker.
func (pool *workflowPoolImpl) Destroy(ctx context.Context) error {
pool.mu.Lock()
defer pool.mu.Unlock()
const op = errors.Op("workflow_pool_destroy")
pool.active = false
for i := 0; i < len(pool.tWorkers); i++ {
pool.tWorkers[i].Stop()
}
worker.PurgeStickyWorkflowCache()
err := pool.worker.Stop()
if err != nil {
return errors.E(op, err)
}
return nil
}
// NewWorkflowDefinition initiates new workflow process.
func (pool *workflowPoolImpl) NewWorkflowDefinition() bindings.WorkflowDefinition {
return &workflowProcess{
codec: pool.codec,
pool: pool,
}
}
// NewWorkflowDefinition initiates new workflow process.
func (pool *workflowPoolImpl) SeqID() uint64 {
return atomic.AddUint64(&pool.seqID, 1)
}
// Exec set of commands in thread safe move.
func (pool *workflowPoolImpl) Exec(p payload.Payload) (payload.Payload, error) {
pool.mu.Lock()
defer pool.mu.Unlock()
if !pool.active {
return payload.Payload{}, nil
}
return pool.worker.Exec(p)
}
func (pool *workflowPoolImpl) Workers() []rrWorker.BaseProcess {
return []rrWorker.BaseProcess{pool.worker}
}
func (pool *workflowPoolImpl) WorkflowNames() []string {
names := make([]string, 0, len(pool.workflows))
for name := range pool.workflows {
names = append(names, name)
}
return names
}
// initWorkers request workers workflows from underlying PHP and configures temporal workers linked to the pool.
func (pool *workflowPoolImpl) initWorkers(ctx context.Context, temporal client.Temporal) error {
const op = errors.Op("workflow_pool_init_workers")
workerInfo, err := rrt.FetchWorkerInfo(pool.codec, pool, temporal.GetDataConverter())
if err != nil {
return errors.E(op, err)
}
pool.workflows = make(map[string]rrt.WorkflowInfo)
pool.tWorkers = make([]worker.Worker, 0)
for _, info := range workerInfo {
w, err := temporal.CreateWorker(info.TaskQueue, info.Options)
if err != nil {
return errors.E(op, err, pool.Destroy(ctx))
}
pool.tWorkers = append(pool.tWorkers, w)
for _, workflowInfo := range info.Workflows {
w.RegisterWorkflowWithOptions(pool, workflow.RegisterOptions{
Name: workflowInfo.Name,
DisableAlreadyRegisteredCheck: false,
})
pool.workflows[workflowInfo.Name] = workflowInfo
}
}
return nil
}
|
