go/go-micro
1
0
Fork 0
mirror of https://github.com/go-micro/go-micro.git synced 2025-05-31 21:59:42 +02:00
Code Issues Releases Activity

fix: default memory registry, add registrations for mdns, nats

Browse Source
This commit is contained in:
Brian Ketelsen 2025-05-07 16:26:29 -04:00
parent 32942a3f63
commit 04bbbf051a
15 changed files with 162 additions and 174 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
cmd/cmd.go
View File

@ -19,6 +19,8 @@ import (
"go-micro.dev/v5/debug/trace"
"go-micro.dev/v5/logger"
"go-micro.dev/v5/registry"
"go-micro.dev/v5/registry/mdns"
"go-micro.dev/v5/registry/nats"
"go-micro.dev/v5/selector"
"go-micro.dev/v5/server"
"go-micro.dev/v5/store"
@ -232,7 +234,11 @@ var (
DefaultClients = map[string]func(...client.Option) client.Client{}
DefaultRegistries = map[string]func(...registry.Option) registry.Registry{}
DefaultRegistries = map[string]func(...registry.Option) registry.Registry{
"memory": registry.NewMemoryRegistry,
"nats": nats.NewRegistry,
"mdns": mdns.NewMDNSRegistry,
}
DefaultSelectors = map[string]func(...selector.Option) selector.Selector{}

8
registry/mdns.go
View File

@ -1,8 +0,0 @@
//go:build !nats
// +build !nats
package registry
var (
DefaultRegistry = NewRegistry()
)

5
registry/mdns/mdns.go Normal file
View File

@ -0,0 +1,5 @@
package mdns
var (
DefaultRegistry = NewMDNSRegistry()
)

58
registry/mdns_registry.go → registry/mdns/mdns_registry.go
View File

@ -1,8 +1,5 @@
//go:build !nats
// +build !nats
// Package mdns is a multicast dns registry
package registry
package mdns
import (
"bytes"
@ -20,6 +17,7 @@ import (
"github.com/google/uuid"
log "go-micro.dev/v5/logger"
"go-micro.dev/v5/registry"
"go-micro.dev/v5/util/mdns"
)
@ -32,7 +30,7 @@ type mdnsTxt struct {
Metadata map[string]string
Service string
Version string
Endpoints []*Endpoint
Endpoints []*registry.Endpoint
}
type mdnsEntry struct {
@ -41,7 +39,7 @@ type mdnsEntry struct {
}
type mdnsRegistry struct {
opts *Options
opts *registry.Options
services map[string][]*mdnsEntry
// watchers
@ -58,7 +56,7 @@ type mdnsRegistry struct {
}
type mdnsWatcher struct {
wo WatchOptions
wo registry.WatchOptions
ch chan *mdns.ServiceEntry
exit chan struct{}
// the registry
@ -130,9 +128,9 @@ func decode(record []string) (*mdnsTxt, error) {
return txt, nil
}
func newRegistry(opts ...Option) Registry {
mergedOpts := append([]Option{Timeout(time.Millisecond * 100)}, opts...)
options := NewOptions(mergedOpts...)
func newRegistry(opts ...registry.Option) registry.Registry {
mergedOpts := append([]registry.Option{registry.Timeout(time.Millisecond * 100)}, opts...)
options := registry.NewOptions(mergedOpts...)
// set the domain
domain := mdnsDomain
@ -150,18 +148,18 @@ func newRegistry(opts ...Option) Registry {
}
}
func (m *mdnsRegistry) Init(opts ...Option) error {
func (m *mdnsRegistry) Init(opts ...registry.Option) error {
for _, o := range opts {
o(m.opts)
}
return nil
}
func (m *mdnsRegistry) Options() Options {
func (m *mdnsRegistry) Options() registry.Options {
return *m.opts
}
func (m *mdnsRegistry) Register(service *Service, opts ...RegisterOption) error {
func (m *mdnsRegistry) Register(service *registry.Service, opts ...registry.RegisterOption) error {
m.Lock()
defer m.Unlock()
@ -266,7 +264,7 @@ func (m *mdnsRegistry) Register(service *Service, opts ...RegisterOption) error
return gerr
}
func (m *mdnsRegistry) Deregister(service *Service, opts ...DeregisterOption) error {
func (m *mdnsRegistry) Deregister(service *registry.Service, opts ...registry.DeregisterOption) error {
m.Lock()
defer m.Unlock()
@ -301,9 +299,9 @@ func (m *mdnsRegistry) Deregister(service *Service, opts ...DeregisterOption) er
return nil
}
func (m *mdnsRegistry) GetService(service string, opts ...GetOption) ([]*Service, error) {
func (m *mdnsRegistry) GetService(service string, opts ...registry.GetOption) ([]*registry.Service, error) {
logger := m.opts.Logger
serviceMap := make(map[string]*Service)
serviceMap := make(map[string]*registry.Service)
entries := make(chan *mdns.ServiceEntry, 10)
done := make(chan bool)
@ -343,7 +341,7 @@ func (m *mdnsRegistry) GetService(service string, opts ...GetOption) ([]*Service
s, ok := serviceMap[txt.Version]
if !ok {
s = &Service{
s = &registry.Service{
Name: txt.Service,
Version: txt.Version,
Endpoints: txt.Endpoints,
@ -360,7 +358,7 @@ func (m *mdnsRegistry) GetService(service string, opts ...GetOption) ([]*Service
logger.Logf(log.InfoLevel, "[mdns]: invalid endpoint received: %v", e)
continue
}
s.Nodes = append(s.Nodes, &Node{
s.Nodes = append(s.Nodes, &registry.Node{
Id: strings.TrimSuffix(e.Name, "."+p.Service+"."+p.Domain+"."),
Address: addr,
Metadata: txt.Metadata,
@ -383,7 +381,7 @@ func (m *mdnsRegistry) GetService(service string, opts ...GetOption) ([]*Service
<-done
// create list and return
services := make([]*Service, 0, len(serviceMap))
services := make([]*registry.Service, 0, len(serviceMap))
for _, service := range serviceMap {
services = append(services, service)
@ -392,7 +390,7 @@ func (m *mdnsRegistry) GetService(service string, opts ...GetOption) ([]*Service
return services, nil
}
func (m *mdnsRegistry) ListServices(opts ...ListOption) ([]*Service, error) {
func (m *mdnsRegistry) ListServices(opts ...registry.ListOption) ([]*registry.Service, error) {
serviceMap := make(map[string]bool)
entries := make(chan *mdns.ServiceEntry, 10)
done := make(chan bool)
@ -407,7 +405,7 @@ func (m *mdnsRegistry) ListServices(opts ...ListOption) ([]*Service, error) {
// set domain
p.Domain = m.domain
var services []*Service
var services []*registry.Service
go func() {
for {
@ -422,7 +420,7 @@ func (m *mdnsRegistry) ListServices(opts ...ListOption) ([]*Service, error) {
name := strings.TrimSuffix(e.Name, "."+p.Service+"."+p.Domain+".")
if !serviceMap[name] {
serviceMap[name] = true
services = append(services, &Service{Name: name})
services = append(services, &registry.Service{Name: name})
}
case <-p.Context.Done():
close(done)
@ -442,8 +440,8 @@ func (m *mdnsRegistry) ListServices(opts ...ListOption) ([]*Service, error) {
return services, nil
}
func (m *mdnsRegistry) Watch(opts ...WatchOption) (Watcher, error) {
var wo WatchOptions
func (m *mdnsRegistry) Watch(opts ...registry.WatchOption) (registry.Watcher, error) {
var wo registry.WatchOptions
for _, o := range opts {
o(&wo)
}
@ -540,7 +538,7 @@ func (m *mdnsRegistry) String() string {
return "mdns"
}
func (m *mdnsWatcher) Next() (*Result, error) {
func (m *mdnsWatcher) Next() (*registry.Result, error) {
for {
select {
case e := <-m.ch:
@ -565,7 +563,7 @@ func (m *mdnsWatcher) Next() (*Result, error) {
action = "create"
}
service := &Service{
service := &registry.Service{
Name: txt.Service,
Version: txt.Version,
Endpoints: txt.Endpoints,
@ -586,18 +584,18 @@ func (m *mdnsWatcher) Next() (*Result, error) {
addr = e.Addr.String()
}
service.Nodes = append(service.Nodes, &Node{
service.Nodes = append(service.Nodes, &registry.Node{
Id: strings.TrimSuffix(e.Name, suffix),
Address: addr,
Metadata: txt.Metadata,
})
return &Result{
return &registry.Result{
Action: action,
Service: service,
}, nil
case <-m.exit:
return nil, ErrWatcherStopped
return nil, registry.ErrWatcherStopped
}
}
}
@ -616,6 +614,6 @@ func (m *mdnsWatcher) Stop() {
}
// NewRegistry returns a new default registry which is mdns.
func NewRegistry(opts ...Option) Registry {
func NewMDNSRegistry(opts ...registry.Option) registry.Registry {
return newRegistry(opts...)
}

47
registry/mdns_test.go → registry/mdns/mdns_test.go
View File

@ -1,12 +1,11 @@
//go:build !nats
// +build !nats
package registry
package mdns
import (
"os"
"testing"
"time"
"go-micro.dev/v5/registry"
)
func TestMDNS(t *testing.T) {
@ -15,11 +14,11 @@ func TestMDNS(t *testing.T) {
t.Skip()
}
testData := []*Service{
testData := []*registry.Service{
{
Name: "test1",
Version: "1.0.1",
Nodes: []*Node{
Nodes: []*registry.Node{
{
Id: "test1-1",
Address: "10.0.0.1:10001",
@ -32,7 +31,7 @@ func TestMDNS(t *testing.T) {
{
Name: "test2",
Version: "1.0.2",
Nodes: []*Node{
Nodes: []*registry.Node{
{
Id: "test2-1",
Address: "10.0.0.2:10002",
@ -45,7 +44,7 @@ func TestMDNS(t *testing.T) {
{
Name: "test3",
Version: "1.0.3",
Nodes: []*Node{
Nodes: []*registry.Node{
{
Id: "test3-1",
Address: "10.0.0.3:10003",
@ -58,7 +57,7 @@ func TestMDNS(t *testing.T) {
{
Name: "test4",
Version: "1.0.4",
Nodes: []*Node{
Nodes: []*registry.Node{
{
Id: "test4-1",
Address: "[::]:10004",
@ -72,14 +71,14 @@ func TestMDNS(t *testing.T) {
travis := os.Getenv("TRAVIS")
var opts []Option
var opts []registry.Option
if travis == "true" {
opts = append(opts, Timeout(time.Millisecond*100))
opts = append(opts, registry.Timeout(time.Millisecond*100))
}
// new registry
r := NewRegistry(opts...)
r := NewMDNSRegistry(opts...)
for _, service := range testData {
// register service
@ -159,14 +158,14 @@ func TestEncoding(t *testing.T) {
Metadata: map[string]string{
"foo": "bar",
},
Endpoints: []*Endpoint{
Endpoints: []*registry.Endpoint{
{
Name: "endpoint1",
Request: &Value{
Request: &registry.Value{
Name: "request",
Type: "request",
},
Response: &Value{
Response: &registry.Value{
Name: "response",
Type: "response",
},
@ -216,11 +215,11 @@ func TestWatcher(t *testing.T) {
t.Skip()
}
testData := []*Service{
testData := []*registry.Service{
{
Name: "test1",
Version: "1.0.1",
Nodes: []*Node{
Nodes: []*registry.Node{
{
Id: "test1-1",
Address: "10.0.0.1:10001",
@ -233,7 +232,7 @@ func TestWatcher(t *testing.T) {
{
Name: "test2",
Version: "1.0.2",
Nodes: []*Node{
Nodes: []*registry.Node{
{
Id: "test2-1",
Address: "10.0.0.2:10002",
@ -246,7 +245,7 @@ func TestWatcher(t *testing.T) {
{
Name: "test3",
Version: "1.0.3",
Nodes: []*Node{
Nodes: []*registry.Node{
{
Id: "test3-1",
Address: "10.0.0.3:10003",
@ -259,7 +258,7 @@ func TestWatcher(t *testing.T) {
{
Name: "test4",
Version: "1.0.4",
Nodes: []*Node{
Nodes: []*registry.Node{
{
Id: "test4-1",
Address: "[::]:10004",
@ -271,7 +270,7 @@ func TestWatcher(t *testing.T) {
},
}
testFn := func(service, s *Service) {
testFn := func(service, s *registry.Service) {
if s == nil {
t.Fatalf("Expected one result for %s got nil", service.Name)
}
@ -301,14 +300,14 @@ func TestWatcher(t *testing.T) {
travis := os.Getenv("TRAVIS")
var opts []Option
var opts []registry.Option
if travis == "true" {
opts = append(opts, Timeout(time.Millisecond*100))
opts = append(opts, registry.Timeout(time.Millisecond*100))
}
// new registry
r := NewRegistry(opts...)
r := NewMDNSRegistry(opts...)
w, err := r.Watch()
if err != nil {

72
registry/nats.go → registry/nats/nats.go
View File

@ -1,8 +1,5 @@
//go:build nats
// +build nats
// Package nats provides a NATS registry using broadcast queries
package registry
package nats
import (
"context"
@ -12,11 +9,12 @@ import (
"time"
"github.com/nats-io/nats.go"
"go-micro.dev/v5/registry"
)
type natsRegistry struct {
addrs []string
opts Options
opts registry.Options
nopts nats.Options
queryTopic string
watchTopic string
@ -24,7 +22,7 @@ type natsRegistry struct {
sync.RWMutex
conn *nats.Conn
services map[string][]*Service
services map[string][]*registry.Service
listeners map[string]chan bool
}
@ -34,7 +32,7 @@ var (
defaultRegisterAction = "create"
)
func configure(n *natsRegistry, opts ...Option) error {
func configure(n *natsRegistry, opts ...registry.Option) error {
for _, o := range opts {
o(&n.opts)
}
@ -135,7 +133,7 @@ func (n *natsRegistry) getConn() (*nats.Conn, error) {
return n.conn, nil
}
func (n *natsRegistry) register(s *Service) error {
func (n *natsRegistry) register(s *registry.Service) error {
conn, err := n.getConn()
if err != nil {
return err
@ -145,7 +143,7 @@ func (n *natsRegistry) register(s *Service) error {
defer n.Unlock()
// cache service
n.services[s.Name] = addServices(n.services[s.Name], cp([]*Service{s}))
n.services[s.Name] = addServices(n.services[s.Name], cp([]*registry.Service{s}))
// create query listener
if n.listeners[s.Name] == nil {
@ -153,13 +151,13 @@ func (n *natsRegistry) register(s *Service) error {
// create a subscriber that responds to queries
sub, err := conn.Subscribe(n.queryTopic, func(m *nats.Msg) {
var result *Result
var result *registry.Result
if err := json.Unmarshal(m.Data, &result); err != nil {
return
}
var services []*Service
var services []*registry.Service
switch result.Action {
// is this a get query and we own the service?
@ -207,11 +205,11 @@ func (n *natsRegistry) register(s *Service) error {
return nil
}
func (n *natsRegistry) deregister(s *Service) error {
func (n *natsRegistry) deregister(s *registry.Service) error {
n.Lock()
defer n.Unlock()
services := delServices(n.services[s.Name], cp([]*Service{s}))
services := delServices(n.services[s.Name], cp([]*registry.Service{s}))
if len(services) > 0 {
n.services[s.Name] = services
return nil
@ -229,28 +227,28 @@ func (n *natsRegistry) deregister(s *Service) error {
return nil
}
func (n *natsRegistry) query(s string, quorum int) ([]*Service, error) {
func (n *natsRegistry) query(s string, quorum int) ([]*registry.Service, error) {
conn, err := n.getConn()
if err != nil {
return nil, err
}
var action string
var service *Service
var service *registry.Service
if len(s) > 0 {
action = "get"
service = &Service{Name: s}
service = &registry.Service{Name: s}
} else {
action = "list"
}
inbox := nats.NewInbox()
response := make(chan *Service, 10)
response := make(chan *registry.Service, 10)
sub, err := conn.Subscribe(inbox, func(m *nats.Msg) {
var service *Service
var service *registry.Service
if err := json.Unmarshal(m.Data, &service); err != nil {
return
}
@ -264,7 +262,7 @@ func (n *natsRegistry) query(s string, quorum int) ([]*Service, error) {
}
defer sub.Unsubscribe()
b, err := json.Marshal(&Result{Action: action, Service: service})
b, err := json.Marshal(&registry.Result{Action: action, Service: service})
if err != nil {
return nil, err
}
@ -279,7 +277,7 @@ func (n *natsRegistry) query(s string, quorum int) ([]*Service, error) {
timeoutChan := time.After(n.opts.Timeout)
serviceMap := make(map[string]*Service)
serviceMap := make(map[string]*registry.Service)
loop:
for {
@ -302,22 +300,22 @@ loop:
}
}
var services []*Service
var services []*registry.Service
for _, service := range serviceMap {
services = append(services, service)
}
return services, nil
}
func (n *natsRegistry) Init(opts ...Option) error {
func (n *natsRegistry) Init(opts ...registry.Option) error {
return configure(n, opts...)
}
func (n *natsRegistry) Options() Options {
func (n *natsRegistry) Options() registry.Options {
return n.opts
}
func (n *natsRegistry) Register(s *Service, opts ...RegisterOption) error {
func (n *natsRegistry) Register(s *registry.Service, opts ...registry.RegisterOption) error {
if err := n.register(s); err != nil {
return err
}
@ -327,7 +325,7 @@ func (n *natsRegistry) Register(s *Service, opts ...RegisterOption) error {
return err
}
b, err := json.Marshal(&Result{Action: n.registerAction, Service: s})
b, err := json.Marshal(&registry.Result{Action: n.registerAction, Service: s})
if err != nil {
return err
}
@ -335,7 +333,7 @@ func (n *natsRegistry) Register(s *Service, opts ...RegisterOption) error {
return conn.Publish(n.watchTopic, b)
}
func (n *natsRegistry) Deregister(s *Service, opts ...DeregisterOption) error {
func (n *natsRegistry) Deregister(s *registry.Service, opts ...registry.DeregisterOption) error {
if err := n.deregister(s); err != nil {
return err
}
@ -345,14 +343,14 @@ func (n *natsRegistry) Deregister(s *Service, opts ...DeregisterOption) error {
return err
}
b, err := json.Marshal(&Result{Action: "delete", Service: s})
b, err := json.Marshal(&registry.Result{Action: "delete", Service: s})
if err != nil {
return err
}
return conn.Publish(n.watchTopic, b)
}
func (n *natsRegistry) GetService(s string, opts ...GetOption) ([]*Service, error) {
func (n *natsRegistry) GetService(s string, opts ...registry.GetOption) ([]*registry.Service, error) {
services, err := n.query(s, getQuorum(n.opts))
if err != nil {
return nil, err
@ -360,17 +358,17 @@ func (n *natsRegistry) GetService(s string, opts ...GetOption) ([]*Service, erro
return services, nil
}
func (n *natsRegistry) ListServices(opts ...ListOption) ([]*Service, error) {
func (n *natsRegistry) ListServices(opts ...registry.ListOption) ([]*registry.Service, error) {
s, err := n.query("", 0)
if err != nil {
return nil, err
}
var services []*Service
serviceMap := make(map[string]*Service)
var services []*registry.Service
serviceMap := make(map[string]*registry.Service)
for _, v := range s {
serviceMap[v.Name] = &Service{Name: v.Name, Version: v.Version}
serviceMap[v.Name] = &registry.Service{Name: v.Name, Version: v.Version}
}
for _, v := range serviceMap {
@ -380,7 +378,7 @@ func (n *natsRegistry) ListServices(opts ...ListOption) ([]*Service, error) {
return services, nil
}
func (n *natsRegistry) Watch(opts ...WatchOption) (Watcher, error) {
func (n *natsRegistry) Watch(opts ...registry.WatchOption) (registry.Watcher, error) {
conn, err := n.getConn()
if err != nil {
return nil, err
@ -391,7 +389,7 @@ func (n *natsRegistry) Watch(opts ...WatchOption) (Watcher, error) {
return nil, err
}
var wo WatchOptions
var wo registry.WatchOptions
for _, o := range opts {
o(&wo)
}
@ -403,15 +401,15 @@ func (n *natsRegistry) String() string {
return "nats"
}
func NewRegistry(opts ...Option) Registry {
options := Options{
func NewRegistry(opts ...registry.Option) registry.Registry {
options := registry.Options{
Timeout: time.Millisecond * 100,
Context: context.Background(),
}
n := &natsRegistry{
opts: options,
services: make(map[string][]*Service),
services: make(map[string][]*registry.Service),
listeners: make(map[string]chan bool),
}
configure(n, opts...)

5
registry/nats_assert_test.go → registry/nats/nats_assert_test.go
View File

@ -1,7 +1,4 @@
//go:build nats
// +build nats
package registry_test
package nats_test
import (
"reflect"

32
registry/nats_environment_test.go → registry/nats/nats_environment_test.go
View File

@ -1,7 +1,4 @@
//go:build nats
// +build nats
package registry_test
package nats_test
import (
"os"
@ -9,19 +6,20 @@ import (
log "go-micro.dev/v5/logger"
"go-micro.dev/v5/registry"
"go-micro.dev/v5/registry/nats"
)
type environment struct {
registryOne Registry
registryTwo Registry
registryThree Registry
registryOne registry.Registry
registryTwo registry.Registry
registryThree registry.Registry
serviceOne Service
serviceTwo Service
serviceOne registry.Service
serviceTwo registry.Service
nodeOne Node
nodeTwo Node
nodeThree Node
nodeOne registry.Node
nodeTwo registry.Node
nodeThree registry.Node
}
var e environment
@ -33,17 +31,17 @@ func TestMain(m *testing.M) {
return
}
e.registryOne = registry.NewRegistry(Addrs(natsURL), registry.Quorum(1))
e.registryTwo = registry.NewRegistry(Addrs(natsURL), registry.Quorum(1))
e.registryThree = registry.NewRegistry(Addrs(natsURL), registry.Quorum(1))
e.registryOne = nats.NewRegistry(registry.Addrs(natsURL), nats.Quorum(1))
e.registryTwo = nats.NewRegistry(registry.Addrs(natsURL), nats.Quorum(1))
e.registryThree = nats.NewRegistry(registry.Addrs(natsURL), nats.Quorum(1))
e.serviceOne.Name = "one"
e.serviceOne.Version = "default"
e.serviceOne.Nodes = []*Node{&e.nodeOne}
e.serviceOne.Nodes = []*registry.Node{&e.nodeOne}
e.serviceTwo.Name = "two"
e.serviceTwo.Version = "default"
e.serviceTwo.Nodes = []*Node{&e.nodeOne, &e.nodeTwo}
e.serviceTwo.Nodes = []*registry.Node{&e.nodeOne, &e.nodeTwo}
e.nodeOne.Id = "one"
e.nodeTwo.Id = "two"

28
registry/nats_options.go → registry/nats/nats_options.go
View File

@ -1,12 +1,10 @@
//go:build nats
// +build nats
package registry
package nats
import (
"context"
"github.com/nats-io/nats.go"
"go-micro.dev/v5/registry"
)
type contextQuorumKey struct{}
@ -19,7 +17,7 @@ var (
DefaultQuorum = 0
)
func getQuorum(o Options) int {
func getQuorum(o registry.Options) int {
if o.Context == nil {
return DefaultQuorum
}
@ -32,16 +30,16 @@ func getQuorum(o Options) int {
}
}
func Quorum(n int) Option {
return func(o *Options) {
func Quorum(n int) registry.Option {
return func(o *registry.Options) {
o.Context = context.WithValue(o.Context, contextQuorumKey{}, n)
}
}
// Options allow to inject a nats.Options struct for configuring
// the nats connection.
func NatsOptions(nopts nats.Options) Option {
return func(o *Options) {
func NatsOptions(nopts nats.Options) registry.Option {
return func(o *registry.Options) {
if o.Context == nil {
o.Context = context.Background()
}
@ -52,8 +50,8 @@ func NatsOptions(nopts nats.Options) Option {
// QueryTopic allows to set a custom nats topic on which service registries
// query (survey) other services. All registries listen on this topic and
// then respond to the query message.
func QueryTopic(s string) Option {
return func(o *Options) {
func QueryTopic(s string) registry.Option {
return func(o *registry.Options) {
if o.Context == nil {
o.Context = context.Background()
}
@ -65,8 +63,8 @@ func QueryTopic(s string) Option {
// changes (e.g. when services are added, updated or removed). Since we don't
// have a central registry service, each service typically broadcasts in a
// determined frequency on this topic.
func WatchTopic(s string) Option {
return func(o *Options) {
func WatchTopic(s string) registry.Option {
return func(o *registry.Options) {
if o.Context == nil {
o.Context = context.Background()
}
@ -79,8 +77,8 @@ func WatchTopic(s string) Option {
// - "create" (default) only registers if there is noone already registered under the same key.
// - "update" only updates the registration if it already exists.
// - "put" creates or updates a registration
func RegisterAction(s string) Option {
return func(o *Options) {
func RegisterAction(s string) registry.Option {
return func(o *registry.Options) {
if o.Context == nil {
o.Context = context.Background()
}

5
registry/nats/nats_registry.go Normal file
View File

@ -0,0 +1,5 @@
package nats
var (
DefaultRegistry = NewRegistry()
)

13
registry/nats_test.go → registry/nats/nats_test.go
View File

@ -1,14 +1,13 @@
//go:build nats
// +build nats
package registry_test
package nats_test
import (
"testing"
"go-micro.dev/v5/registry"
)
func TestRegister(t *testing.T) {
service := Service{Name: "test"}
service := registry.Service{Name: "test"}
assertNoError(t, e.registryOne.Register(&service))
defer e.registryOne.Deregister(&service)
@ -22,8 +21,8 @@ func TestRegister(t *testing.T) {
}
func TestDeregister(t *testing.T) {
service1 := Service{Name: "test-deregister", Version: "v1"}
service2 := Service{Name: "test-deregister", Version: "v2"}
service1 := registry.Service{Name: "test-deregister", Version: "v1"}
service2 := registry.Service{Name: "test-deregister", Version: "v2"}
assertNoError(t, e.registryOne.Register(&service1))
services, err := e.registryOne.GetService(service1.Name)

31
registry/nats_util.go → registry/nats/nats_util.go
View File

@ -1,29 +1,28 @@
//go:build nats
// +build nats
package nats
package registry
import "go-micro.dev/v5/registry"
func cp(current []*Service) []*Service {
var services []*Service
func cp(current []*registry.Service) []*registry.Service {
var services []*registry.Service
for _, service := range current {
// copy service
s := new(Service)
s := new(registry.Service)
*s = *service
// copy nodes
var nodes []*Node
var nodes []*registry.Node
for _, node := range service.Nodes {
n := new(Node)
n := new(registry.Node)
*n = *node
nodes = append(nodes, n)
}
s.Nodes = nodes
// copy endpoints
var eps []*Endpoint
var eps []*registry.Endpoint
for _, ep := range service.Endpoints {
e := new(Endpoint)
e := new(registry.Endpoint)
*e = *ep
eps = append(eps, e)
}
@ -36,7 +35,7 @@ func cp(current []*Service) []*Service {
return services
}
func addNodes(old, neu []*Node) []*Node {
func addNodes(old, neu []*registry.Node) []*registry.Node {
for _, n := range neu {
var seen bool
for i, o := range old {
@ -53,7 +52,7 @@ func addNodes(old, neu []*Node) []*Node {
return old
}
func addServices(old, neu []*Service) []*Service {
func addServices(old, neu []*registry.Service) []*registry.Service {
for _, s := range neu {
var seen bool
for i, o := range old {
@ -71,8 +70,8 @@ func addServices(old, neu []*Service) []*Service {
return old
}
func delNodes(old, del []*Node) []*Node {
var nodes []*Node
func delNodes(old, del []*registry.Node) []*registry.Node {
var nodes []*registry.Node
for _, o := range old {
var rem bool
for _, n := range del {
@ -88,8 +87,8 @@ func delNodes(old, del []*Node) []*Node {
return nodes
}
func delServices(old, del []*Service) []*Service {
var services []*Service
func delServices(old, del []*registry.Service) []*registry.Service {
var services []*registry.Service
for i, o := range old {
var rem bool
for _, s := range del {

12
registry/nats_watcher.go → registry/nats/nats_watcher.go
View File

@ -1,22 +1,20 @@
//go:build nats
// +build nats
package registry
package nats
import (
"encoding/json"
"time"
"github.com/nats-io/nats.go"
"go-micro.dev/v5/registry"
)
type natsWatcher struct {
sub *nats.Subscription
wo WatchOptions
wo registry.WatchOptions
}
func (n *natsWatcher) Next() (*Result, error) {
var result *Result
func (n *natsWatcher) Next() (*registry.Result, error) {
var result *registry.Result
for {
m, err := n.sub.NextMsg(time.Minute)
if err != nil && err == nats.ErrTimeout {

8
registry/nats_registry.go
View File

@ -1,8 +0,0 @@
//go:build nats
// +build nats
package registry
var (
DefaultRegistry = NewRegistry()
)

4
registry/registry.go
View File

@ -93,3 +93,7 @@ func Watch(opts ...WatchOption) (Watcher, error) {
func String() string {
return DefaultRegistry.String()
}
var (
DefaultRegistry = NewMemoryRegistry()
)