unistack-org/micro
4
0
Fork 2
Code Issues 18 Pull Requests Actions Packages Projects Releases 324 Wiki Activity

Add context options to the runtime

Browse Source
This commit is contained in:
Ben Toogood
2020-04-14 12:32:59 +01:00
parent 0c75a0306b
commit e17825474f
10 changed files with 112 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
registry/cache/cache.go vendored
View File

@@ -419,7 +419,7 @@ func (c *cache) watch(w registry.Watcher) error {
}
}
func (c *cache) GetService(service string) ([]*registry.Service, error) {
func (c *cache) GetService(service string, opts ...registry.GetOption) ([]*registry.Service, error) {
// get the service
services, err := c.get(service)
if err != nil {

6
registry/etcd/etcd.go
View File

@@ -276,7 +276,7 @@ func (e *etcdRegistry) registerNode(s *registry.Service, node *registry.Node, op
return nil
}
func (e *etcdRegistry) Deregister(s *registry.Service) error {
func (e *etcdRegistry) Deregister(s *registry.Service, opts ...registry.DeregisterOption) error {
if len(s.Nodes) == 0 {
return errors.New("Require at least one node")
}
@@ -322,7 +322,7 @@ func (e *etcdRegistry) Register(s *registry.Service, opts ...registry.RegisterOp
return gerr
}
func (e *etcdRegistry) GetService(name string) ([]*registry.Service, error) {
func (e *etcdRegistry) GetService(name string, opts ...registry.GetOption) ([]*registry.Service, error) {
ctx, cancel := context.WithTimeout(context.Background(), e.options.Timeout)
defer cancel()
@@ -362,7 +362,7 @@ func (e *etcdRegistry) GetService(name string) ([]*registry.Service, error) {
return services, nil
}
func (e *etcdRegistry) ListServices() ([]*registry.Service, error) {
func (e *etcdRegistry) ListServices(opts ...registry.ListOption) ([]*registry.Service, error) {
versions := make(map[string]*registry.Service)
ctx, cancel := context.WithTimeout(context.Background(), e.options.Timeout)

6
registry/mdns_registry.go
View File

@@ -269,7 +269,7 @@ func (m *mdnsRegistry) Register(service *Service, opts ...RegisterOption) error
return gerr
}
func (m *mdnsRegistry) Deregister(service *Service) error {
func (m *mdnsRegistry) Deregister(service *Service, opts ...DeregisterOption) error {
m.Lock()
defer m.Unlock()
@@ -304,7 +304,7 @@ func (m *mdnsRegistry) Deregister(service *Service) error {
return nil
}
func (m *mdnsRegistry) GetService(service string) ([]*Service, error) {
func (m *mdnsRegistry) GetService(service string, opts ...GetOption) ([]*Service, error) {
serviceMap := make(map[string]*Service)
entries := make(chan *mdns.ServiceEntry, 10)
done := make(chan bool)
@@ -396,7 +396,7 @@ func (m *mdnsRegistry) GetService(service string) ([]*Service, error) {
return services, nil
}
func (m *mdnsRegistry) ListServices() ([]*Service, error) {
func (m *mdnsRegistry) ListServices(opts ...ListOption) ([]*Service, error) {
serviceMap := make(map[string]bool)
entries := make(chan *mdns.ServiceEntry, 10)
done := make(chan bool)

6
registry/memory/memory.go
View File

@@ -207,7 +207,7 @@ func (m *Registry) Register(s *registry.Service, opts ...registry.RegisterOption
return nil
}
func (m *Registry) Deregister(s *registry.Service) error {
func (m *Registry) Deregister(s *registry.Service, opts ...registry.DeregisterOption) error {
m.Lock()
defer m.Unlock()
@@ -240,7 +240,7 @@ func (m *Registry) Deregister(s *registry.Service) error {
return nil
}
func (m *Registry) GetService(name string) ([]*registry.Service, error) {
func (m *Registry) GetService(name string, opts ...registry.GetOption) ([]*registry.Service, error) {
m.RLock()
defer m.RUnlock()
@@ -259,7 +259,7 @@ func (m *Registry) GetService(name string) ([]*registry.Service, error) {
return services, nil
}
func (m *Registry) ListServices() ([]*registry.Service, error) {
func (m *Registry) ListServices(opts ...registry.ListOption) ([]*registry.Service, error) {
m.RLock()
defer m.RUnlock()

42
registry/options.go
View File

@@ -32,6 +32,18 @@ type WatchOptions struct {
Context context.Context
}
type DeregisterOptions struct {
Context context.Context
}
type GetOptions struct {
Context context.Context
}
type ListOptions struct {
Context context.Context
}
// Addrs is the registry addresses to use
func Addrs(addrs ...string) Option {
return func(o *Options) {
@@ -65,9 +77,39 @@ func RegisterTTL(t time.Duration) RegisterOption {
}
}
func RegisterContext(ctx context.Context) RegisterOption {
return func(o *RegisterOptions) {
o.Context = ctx
}
}
// Watch a service
func WatchService(name string) WatchOption {
return func(o *WatchOptions) {
o.Service = name
}
}
func WatchContext(ctx context.Context) WatchOption {
return func(o *WatchOptions) {
o.Context = ctx
}
}
func DeregisterContext(ctx context.Context) DeregisterOption {
return func(o *DeregisterOptions) {
o.Context = ctx
}
}
func GetContext(ctx context.Context) GetOption {
return func(o *GetOptions) {
o.Context = ctx
}
}
func ListContext(ctx context.Context) ListOption {
return func(o *ListOptions) {
o.Context = ctx
}
}

12
registry/registry.go
View File

@@ -21,9 +21,9 @@ type Registry interface {
Init(...Option) error
Options() Options
Register(*Service, ...RegisterOption) error
Deregister(*Service) error
GetService(string) ([]*Service, error)
ListServices() ([]*Service, error)
Deregister(*Service, ...DeregisterOption) error
GetService(string, ...GetOption) ([]*Service, error)
ListServices(...ListOption) ([]*Service, error)
Watch(...WatchOption) (Watcher, error)
String() string
}
@@ -61,6 +61,12 @@ type RegisterOption func(*RegisterOptions)
type WatchOption func(*WatchOptions)
type DeregisterOption func(*DeregisterOptions)
type GetOption func(*GetOptions)
type ListOption func(*ListOptions)
// Register a service node. Additionally supply options such as TTL.
func Register(s *Service, opts ...RegisterOption) error {
return DefaultRegistry.Register(s, opts...)

46
registry/service/service.go
View File

@@ -58,13 +58,16 @@ func (s *serviceRegistry) Register(srv *registry.Service, opts ...registry.Regis
for _, o := range opts {
o(&options)
}
if options.Context == nil {
options.Context = context.TODO()
}
// encode srv into protobuf and pack Register TTL into it
pbSrv := ToProto(srv)
pbSrv.Options.Ttl = int64(options.TTL.Seconds())
// register the service
_, err := s.client.Register(context.TODO(), pbSrv, s.callOpts()...)
_, err := s.client.Register(options.Context, pbSrv, s.callOpts()...)
if err != nil {
return err
}
@@ -72,17 +75,33 @@ func (s *serviceRegistry) Register(srv *registry.Service, opts ...registry.Regis
return nil
}
func (s *serviceRegistry) Deregister(srv *registry.Service) error {
func (s *serviceRegistry) Deregister(srv *registry.Service, opts ...registry.DeregisterOption) error {
var options registry.DeregisterOptions
for _, o := range opts {
o(&options)
}
if options.Context == nil {
options.Context = context.TODO()
}
// deregister the service
_, err := s.client.Deregister(context.TODO(), ToProto(srv), s.callOpts()...)
_, err := s.client.Deregister(options.Context, ToProto(srv), s.callOpts()...)
if err != nil {
return err
}
return nil
}
func (s *serviceRegistry) GetService(name string) ([]*registry.Service, error) {
rsp, err := s.client.GetService(context.TODO(), &pb.GetRequest{
func (s *serviceRegistry) GetService(name string, opts ...registry.GetOption) ([]*registry.Service, error) {
var options registry.GetOptions
for _, o := range opts {
o(&options)
}
if options.Context == nil {
options.Context = context.TODO()
}
rsp, err := s.client.GetService(options.Context, &pb.GetRequest{
Service: name,
}, s.callOpts()...)
@@ -97,8 +116,16 @@ func (s *serviceRegistry) GetService(name string) ([]*registry.Service, error) {
return services, nil
}
func (s *serviceRegistry) ListServices() ([]*registry.Service, error) {
rsp, err := s.client.ListServices(context.TODO(), &pb.ListRequest{}, s.callOpts()...)
func (s *serviceRegistry) ListServices(opts ...registry.ListOption) ([]*registry.Service, error) {
var options registry.ListOptions
for _, o := range opts {
o(&options)
}
if options.Context == nil {
options.Context = context.TODO()
}
rsp, err := s.client.ListServices(options.Context, &pb.ListRequest{}, s.callOpts()...)
if err != nil {
return nil, err
}
@@ -116,8 +143,11 @@ func (s *serviceRegistry) Watch(opts ...registry.WatchOption) (registry.Watcher,
for _, o := range opts {
o(&options)
}
if options.Context == nil {
options.Context = context.TODO()
}
stream, err := s.client.Watch(context.TODO(), &pb.WatchRequest{
stream, err := s.client.Watch(options.Context, &pb.WatchRequest{
Service: options.Service,
}, s.callOpts()...)