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
|
package service
import (
"fmt"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"sync"
)
// Config provides ability to slice configuration sections and unmarshal configuration data into
// given structure.
type Config interface {
// Get nested config section (sub-map), returns nil if section not found.
Get(service string) Config
// Unmarshal unmarshal config data into given struct.
Unmarshal(out interface{}) error
}
// Registry controls all internal RR services and provides plugin based system.
type Registry interface {
// Register add new service to the registry under given name.
Register(name string, service Service)
// Configure configures all underlying services with given configuration.
Configure(cfg Config) error
// Check is Service has been registered and configured.
Has(service string) bool
// Get returns Service instance by it's Name or nil if Service not found. Method must return only configured instance.
Get(service string) Service
// Serve all configured services. Non blocking.
Serve() error
// Stop all active services.
Stop() error
}
// Service provides high level functionality for road runner Service.
type Service interface {
// WithConfig must return Service instance configured with the given environment. Must return error in case of
// misconfiguration, might return nil as Service if Service is not enabled.
WithConfig(cfg Config, reg Registry) (Service, error)
// Serve serves Service.
Serve() error
// Stop stop Service Service.
Stop() error
}
type registry struct {
log logrus.FieldLogger
mu sync.Mutex
candidates []*entry
configured []*entry
}
// entry creates association between service instance and given name.
type entry struct {
// Associated service name
Name string
// Associated service instance
Service Service
// Serving indicates that service is currently serving
Serving bool
}
// NewRegistry creates new registry.
func NewRegistry(log logrus.FieldLogger) Registry {
return ®istry{
log: log,
candidates: make([]*entry, 0),
}
}
// Register add new service to the registry under given name.
func (r *registry) Register(name string, service Service) {
r.mu.Lock()
defer r.mu.Unlock()
r.candidates = append(r.candidates, &entry{
Name: name,
Service: service,
Serving: false,
})
r.log.Debugf("%s.service: registered", name)
}
// Configure configures all underlying services with given configuration.
func (r *registry) Configure(cfg Config) error {
if r.configured != nil {
return fmt.Errorf("service bus has been already configured")
}
r.configured = make([]*entry, 0)
for _, e := range r.candidates {
segment := cfg.Get(e.Name)
if segment == nil {
r.log.Debugf("%s.service: no config has been provided", e.Name)
continue
}
s, err := e.Service.WithConfig(segment, r)
if err != nil {
return errors.Wrap(err, fmt.Sprintf("%s.service", e.Name))
}
if s != nil {
r.configured = append(r.configured, &entry{
Name: e.Name,
Service: s,
Serving: false,
})
}
}
return nil
}
// Check is Service has been registered.
func (r *registry) Has(service string) bool {
r.mu.Lock()
defer r.mu.Unlock()
for _, e := range r.configured {
if e.Name == service {
return true
}
}
return false
}
// Get returns Service instance by it's Name or nil if Service not found.
func (r *registry) Get(service string) Service {
r.mu.Lock()
defer r.mu.Unlock()
for _, e := range r.configured {
if e.Name == service {
return e.Service
}
}
return nil
}
// Serve all configured services. Non blocking.
func (r *registry) Serve() error {
return nil
}
// Stop all active services.
func (r *registry) Stop() error {
return nil
}
|
