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
|
package roadrunner
import (
"fmt"
"os/exec"
"sync"
)
const (
// EventNewPool triggered when server creates new pool.
EventNewPool = 60
// EventDestroyPool triggered when server destroys existed pool.
EventDestroyPool = 61
)
// Service manages pool creation and swapping.
type Server struct {
// worker command creator
cmd func() *exec.Cmd
// defines server wide configuration, behaviour and timeouts.
config ServerConfig
// observes pool events (can be attached to multiple pools at the same time)
observer func(event int, ctx interface{})
// creates and connects to workers
factory Factory
// protects pool while the switch
mu sync.Mutex
// pool behaviour
cfg Config
// currently active pool instance
pool Pool
}
// NewServer creates new router. Make sure to call configure before the usage.
func NewServer(cmd func() *exec.Cmd, factory Factory) *Server {
return &Server{
cmd: cmd,
factory: factory,
}
}
// Configure configures underlying pool and destroys it's previous version if any.
func (r *Server) Configure(cfg Config) error {
r.mu.Lock()
previous := r.pool
r.mu.Unlock()
pool, err := NewPool(r.cmd, r.factory, cfg)
if err != nil {
return err
}
r.throw(EventNewPool, pool)
r.mu.Lock()
r.cfg, r.pool = cfg, pool
r.pool.Observe(r.poolObserver)
r.mu.Unlock()
if previous != nil {
go func(p Pool) {
r.throw(EventDestroyPool, p)
p.Destroy()
}(previous)
}
return nil
}
// Reset resets the state of underlying pool and rebuilds all of it's workers.
func (r *Server) Reset() error {
return r.Configure(r.cfg)
}
// Observe attaches event watcher to the router.
func (r *Server) Observe(o func(event int, ctx interface{})) {
r.observer = o
}
// Pool returns active pool or error.
func (r *Server) Pool() Pool {
r.mu.Lock()
defer r.mu.Unlock()
return r.pool
}
// Exec one task with given payload and context, returns result or error.
func (r *Server) Exec(rqs *Payload) (rsp *Payload, err error) {
pool := r.Pool()
if pool == nil {
return nil, fmt.Errorf("no associared pool")
}
return pool.Exec(rqs)
}
// Workers returns worker list associated with the pool.
func (r *Server) Workers() (workers []*Worker) {
pool := r.Pool()
if pool == nil {
return nil
}
return pool.Workers()
}
// Destroy all underlying pools and workers workers (but let them to complete the task).
func (r *Server) Destroy() {
r.mu.Lock()
defer r.mu.Unlock()
if r.pool == nil {
return
}
go func(p Pool) {
r.throw(EventDestroyPool, p)
p.Destroy()
}(r.pool)
r.pool = nil
}
// Start the server underlying worker pool and factory.
func (r *Server) Start() error {
if r.factory != nil {
//todo: already have started
return nil
}
return nil
}
// Stop the server and close underlying factory.
func (r *Server) Stop() {
r.mu.Lock()
defer r.mu.Unlock()
r.factory.Close()
r.factory = nil
}
// throw invokes event handler if any.
func (r *Server) throw(event int, ctx interface{}) {
if r.observer != nil {
r.observer(event, ctx)
}
}
// Observe pool events.
func (r *Server) poolObserver(event int, ctx interface{}) {
// bypassing to user specified observer
r.throw(event, ctx)
if event == EventPoolError {
// pool failure, rebuilding
r.Reset()
}
}
|
