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
|
package roadrunner
import (
"sync"
"os/exec"
"fmt"
)
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 {
// observes pool events (can be attached to multiple pools at the same time)
observer func(event int, ctx interface{})
// worker command creator
cmd func() *exec.Cmd
// pool behaviour
cfg Config
// creates and connects to workers
factory Factory
// protects pool while the switch
mu sync.Mutex
// 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
}
// 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()
}
}
|
