Repository: go-kratos/kratos
Pull request #1986 (2022-05-17 03:35:32)

feat(registry): zookeeper watch node changed
<!--
🎉 Thanks for sending a pull request to Kratos! Here are some tips for you:

1. If this is your first time contributing to Kratos, please read our contribution guide: https://go-kratos.dev/en/docs/community/contribution/
2. Ensure you have added or ran the appropriate tests and lint for your PR, please use `make lint` and `make test` before filing your PR, use `make clean` to tidy your go mod.
3. If the PR is unfinished, you may need mark it as a WIP(Work In Progress) PR or draft PR
4. Please use a conventional commits format title: `<type>[optional scope]: <description>`
    
    Some suggestion on <type>:

    fix: A bug fix
    feat: A new feature
    test: Adding missing tests or correcting existing tests
    refactor: A code change that neither fixes a bug nor adds a feature
    break: Changes has break change

    docs: Documentation only changes
    deps: Changes external dependencies
    style: Changes that do not affect the meaning of the code (white-space, formatting, etc)
    chore Daily work, examples, etc.
    ci: Changes to our CI configuration files and scripts
-->

#### Description (what this PR does / why we need it):
<!--
* The description should include the motivation for this PR or contrast this with previous behavior
-->


#### Which issue(s) this PR fixes (resolves / be part of):
<!--
* Automatically closes linked issue when PR is merged.
* If you PR is not fully resolved issue, please use `part of #<issue number>` instead.

Usage: `fixes/resolves #<issue number>`, or `fixes/resolves (paste link of issue)`.
-->
fixes/zookeeper #1966 

#### Other special notes for reviewer:
<!--
* Somethings that need extra attention for reviewer
* Some additional notes, TODO list, etc.
-->
Zookeeper Registry 的 New 函数签名被重写了

codecov-commenter: # [Codecov](https://codecov.io/gh/go-kratos/kratos/pull/1986?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos) Report
> Merging [#1986](https://codecov.io/gh/go-kratos/kratos/pull/1986?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos) (22a81d7) into [main](https://codecov.io/gh/go-kratos/kratos/commit/ed144f6813c179b7eb0566d5b9ccb6a1377e4a1d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos) (ed144f6) will **increase** coverage by `0.05%`.
> The diff coverage is `n/a`.

```diff
@@            Coverage Diff             @@
##             main    #1986      +/-   ##
==========================================
+ Coverage   80.71%   80.77%   +0.05%     
==========================================
  Files          82       82              
  Lines        3635     3635              
==========================================
+ Hits         2934     2936       +2     
+ Misses        494      492       -2     
  Partials      207      207              
```


| [Impacted Files](https://codecov.io/gh/go-kratos/kratos/pull/1986?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos) | Coverage Δ | |
|---|---|---|
| [internal/context/context.go](https://codecov.io/gh/go-kratos/kratos/pull/1986/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos#diff-aW50ZXJuYWwvY29udGV4dC9jb250ZXh0Lmdv) | `100.00% <0.00%> (+2.98%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/go-kratos/kratos/pull/1986?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/go-kratos/kratos/pull/1986?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos). Last update [ed144f6...22a81d7](https://codecov.io/gh/go-kratos/kratos/pull/1986?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=go-kratos).


Patch:
diff --git a/contrib/registry/zookeeper/go.mod b/contrib/registry/zookeeper/go.mod
index 43bc9349..bb3bb482 100644
--- a/contrib/registry/zookeeper/go.mod
+++ b/contrib/registry/zookeeper/go.mod
@@ -5,6 +5,6 @@ go 1.16
 require (
 	github.com/go-kratos/kratos/v2 v2.2.2
 	github.com/go-zookeeper/zk v1.0.2
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 )
-
 replace github.com/go-kratos/kratos/v2 => ../../../
diff --git a/contrib/registry/zookeeper/go.sum b/contrib/registry/zookeeper/go.sum
index 3b8d2e0f..977a10ca 100644
--- a/contrib/registry/zookeeper/go.sum
+++ b/contrib/registry/zookeeper/go.sum
@@ -26,6 +26,8 @@ github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7
 github.com/fsnotify/fsnotify v1.5.1/go.mod h1:T3375wBYaZdLLcVNkcVbzGHY7f1l/uK5T5Ai1i3InKU=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/go-kratos/aegis v0.1.1/go.mod h1:jYeSQ3Gesba478zEnujOiG5QdsyF3Xk/8owFUeKcHxw=
+github.com/go-kratos/kratos/v2 v2.2.2 h1:5omkfwLiaNnfLdpJkpLwz5beGCehSTDL7vdPQ3lXGtI=
+github.com/go-kratos/kratos/v2 v2.2.2/go.mod h1:yebXu5KMayLjXZzMTY5HWIPRDwcBehHpiNF/Ot8A2pA=
 github.com/go-logr/logr v1.2.0/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/logr v1.2.1/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/stdr v1.2.0/go.mod h1:YkVgnZu1ZjjL7xTxrfm/LLZBfkhTqSR1ydtm6jTKKwI=
@@ -96,6 +98,7 @@ golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
diff --git a/contrib/registry/zookeeper/register.go b/contrib/registry/zookeeper/register.go
index adae0050..2aedc198 100644
--- a/contrib/registry/zookeeper/register.go
+++ b/contrib/registry/zookeeper/register.go
@@ -2,13 +2,10 @@ package zookeeper
 
 import (
 	"context"
-	"encoding/json"
 	"path"
-	"sync"
-	"sync/atomic"
-	"time"
 
 	"github.com/go-zookeeper/zk"
+	"golang.org/x/sync/singleflight"
 
 	"github.com/go-kratos/kratos/v2/registry"
 )
@@ -22,26 +19,14 @@ var (
 type Option func(o *options)
 
 type options struct {
-	ctx      context.Context
-	rootPath string
-	timeout  time.Duration
-	user     string
-	password string
-}
-
-// WithContext with registry context.
-func WithContext(ctx context.Context) Option {
-	return func(o *options) { o.ctx = ctx }
+	namespace string
+	user      string
+	password  string
 }
 
 // WithRootPath with registry root path.
 func WithRootPath(path string) Option {
-	return func(o *options) { o.rootPath = path }
-}
-
-// WithTimeout with registry timeout.
-func WithTimeout(timeout time.Duration) Option {
-	return func(o *options) { o.timeout = timeout }
+	return func(o *options) { o.namespace = path }
 }
 
 // WithDigestACL with registry password.
@@ -54,36 +39,23 @@ func WithDigestACL(user string, password string) Option {
 
 // Registry is consul registry
 type Registry struct {
-	opts     *options
-	conn     *zk.Conn
-	lock     sync.Mutex
-	registry map[string]*serviceSet
+	opts *options
+	conn *zk.Conn
+
+	group singleflight.Group
 }
 
-func New(zkServers []string, opts ...Option) (*Registry, error) {
+func New(conn *zk.Conn, opts ...Option) *Registry {
 	options := &options{
-		ctx:      context.Background(),
-		rootPath: "/microservices",
-		timeout:  time.Second * 5,
+		namespace: "/microservices",
 	}
 	for _, o := range opts {
 		o(options)
 	}
-	conn, _, err := zk.Connect(zkServers, options.timeout)
-	if err != nil {
-		return nil, err
-	}
-	if len(options.user) > 0 && len(options.password) > 0 {
-		err = conn.AddAuth("digest", []byte(options.user+":"+options.password))
-		if err != nil {
-			return nil, err
-		}
-	}
 	return &Registry{
-		opts:     options,
-		conn:     conn,
-		registry: make(map[string]*serviceSet),
-	}, err
+		opts: options,
+		conn: conn,
+	}
 }
 
 func (r *Registry) Register(ctx context.Context, service *registry.ServiceInstance) error {
@@ -91,14 +63,14 @@ func (r *Registry) Register(ctx context.Context, service *registry.ServiceInstan
 		data []byte
 		err  error
 	)
-	if err = r.ensureName(r.opts.rootPath, []byte(""), 0); err != nil {
+	if err = r.ensureName(r.opts.namespace, []byte(""), 0); err != nil {
 		return err
 	}
-	serviceNamePath := path.Join(r.opts.rootPath, service.Name)
+	serviceNamePath := path.Join(r.opts.namespace, service.Name)
 	if err = r.ensureName(serviceNamePath, []byte(""), 0); err != nil {
 		return err
 	}
-	if data, err = json.Marshal(service); err != nil {
+	if data, err = marshal(service); err != nil {
 		return err
 	}
 	servicePath := path.Join(serviceNamePath, service.ID)
@@ -111,7 +83,7 @@ func (r *Registry) Register(ctx context.Context, service *registry.ServiceInstan
 // Deregister registry service to zookeeper.
 func (r *Registry) Deregister(ctx context.Context, service *registry.ServiceInstance) error {
 	ch := make(chan error, 1)
-	servicePath := path.Join(r.opts.rootPath, service.Name, service.ID)
+	servicePath := path.Join(r.opts.namespace, service.Name, service.ID)
 	go func() {
 		err := r.conn.Delete(servicePath, -1)
 		ch <- err
@@ -127,68 +99,36 @@ func (r *Registry) Deregister(ctx context.Context, service *registry.ServiceInst
 
 // GetService get services from zookeeper
 func (r *Registry) GetService(ctx context.Context, serviceName string) ([]*registry.ServiceInstance, error) {
-	serviceNamePath := path.Join(r.opts.rootPath, serviceName)
-	servicesID, _, err := r.conn.Children(serviceNamePath)
-	if err != nil {
-		return nil, err
-	}
-	items := make([]*registry.ServiceInstance, 0, len(servicesID))
-	for _, service := range servicesID {
-		item := &registry.ServiceInstance{}
-		servicePath := path.Join(serviceNamePath, service)
-		serviceInstanceByte, _, err := r.conn.Get(servicePath)
+	instances, err, _ := r.group.Do(serviceName, func() (interface{}, error) {
+		serviceNamePath := path.Join(r.opts.namespace, serviceName)
+		servicesID, _, err := r.conn.Children(serviceNamePath)
 		if err != nil {
 			return nil, err
 		}
-		if err := json.Unmarshal(serviceInstanceByte, item); err != nil {
-			return nil, err
+		items := make([]*registry.ServiceInstance, 0, len(servicesID))
+		for _, service := range servicesID {
+			servicePath := path.Join(serviceNamePath, service)
+			serviceInstanceByte, _, err := r.conn.Get(servicePath)
+			if err != nil {
+				return nil, err
+			}
+			item, err := unmarshal(serviceInstanceByte)
+			if err != nil {
+				return nil, err
+			}
+			items = append(items, item)
 		}
-		items = append(items, item)
+		return items, nil
+	})
+	if err != nil {
+		return nil, err
 	}
-	return items, nil
+	return instances.([]*registry.ServiceInstance), nil
 }
 
 func (r *Registry) Watch(ctx context.Context, serviceName string) (registry.Watcher, error) {
-	r.lock.Lock()
-	defer r.lock.Unlock()
-	set, ok := r.registry[serviceName]
-	if !ok {
-		set = &serviceSet{
-			watcher:     make(map[*watcher]struct{}),
-			services:    &atomic.Value{},
-			serviceName: serviceName,
-		}
-		r.registry[serviceName] = set
-	}
-	// 初始化watcher
-	w := &watcher{
-		event: make(chan struct{}, 1),
-	}
-	w.ctx, w.cancel = context.WithCancel(context.Background())
-	w.set = set
-	set.lock.Lock()
-	set.watcher[w] = struct{}{}
-	set.lock.Unlock()
-	ss, _ := set.services.Load().([]*registry.ServiceInstance)
-	if len(ss) > 0 {
-		// 如果services有值需要推送给watcher，否则watch的时候可能会永远阻塞拿不到初始的数据
-		w.event <- struct{}{}
-	}
-
-	// 放在最后是为了防止漏推送
-	if !ok {
-		go r.resolve(set)
-	}
-	return w, nil
-}
-
-func (r *Registry) resolve(ss *serviceSet) {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
-	services, err := r.GetService(ctx, ss.serviceName)
-	cancel()
-	if err == nil && len(services) > 0 {
-		ss.broadcast(services)
-	}
+	prefix := path.Join(r.opts.namespace, serviceName)
+	return newWatcher(ctx, prefix, serviceName, r.conn)
 }
 
 // ensureName ensure node exists, if not exist, create and set data
diff --git a/contrib/registry/zookeeper/service.go b/contrib/registry/zookeeper/service.go
index 59eb7a13..36cd72be 100644
--- a/contrib/registry/zookeeper/service.go
+++ b/contrib/registry/zookeeper/service.go
@@ -1,27 +1,16 @@
 package zookeeper
 
 import (
-	"sync"
-	"sync/atomic"
+	"encoding/json"
 
 	"github.com/go-kratos/kratos/v2/registry"
 )
 
-type serviceSet struct {
-	serviceName string
-	watcher     map[*watcher]struct{}
-	services    *atomic.Value
-	lock        sync.RWMutex
+func marshal(si *registry.ServiceInstance) ([]byte, error) {
+	return json.Marshal(si)
 }
 
-func (s *serviceSet) broadcast(ss []*registry.ServiceInstance) {
-	s.services.Store(ss)
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	for k := range s.watcher {
-		select {
-		case k.event <- struct{}{}:
-		default:
-		}
-	}
+func unmarshal(data []byte) (si *registry.ServiceInstance, err error) {
+	err = json.Unmarshal(data, &si)
+	return
 }
diff --git a/contrib/registry/zookeeper/watcher.go b/contrib/registry/zookeeper/watcher.go
index c56f3d08..76f502fa 100644
--- a/contrib/registry/zookeeper/watcher.go
+++ b/contrib/registry/zookeeper/watcher.go
@@ -2,36 +2,101 @@ package zookeeper
 
 import (
 	"context"
+	"errors"
+	"path"
+	"sync/atomic"
 
 	"github.com/go-kratos/kratos/v2/registry"
+	"github.com/go-zookeeper/zk"
 )
 
 var _ registry.Watcher = &watcher{}
 
+var ErrWatcherStopped = errors.New("watcher stopped")
+
 type watcher struct {
 	ctx    context.Context
+	event  chan zk.Event
+	conn   *zk.Conn
 	cancel context.CancelFunc
-	event  chan struct{}
-	set    *serviceSet
+
+	first uint32
+	// 前缀
+	prefix string
+	// watch 的服务名
+	serviceName string
+}
+
+func newWatcher(ctx context.Context, prefix, serviceName string, conn *zk.Conn) (*watcher, error) {
+	w := &watcher{conn: conn, event: make(chan zk.Event, 1), prefix: prefix, serviceName: serviceName}
+	w.ctx, w.cancel = context.WithCancel(ctx)
+	go w.watch(w.ctx)
+	return w, nil
 }
 
-func (w watcher) Next() (services []*registry.ServiceInstance, err error) {
+func (w *watcher) watch(ctx context.Context) {
+	for {
+		// 每次 watch 只有一次有效期 所以循环 watch
+		_, _, ch, err := w.conn.ChildrenW(w.prefix)
+		if err != nil {
+			w.event <- zk.Event{Err: err}
+		}
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			w.event <- <-ch
+		}
+	}
+}
+
+func (w *watcher) Next() ([]*registry.ServiceInstance, error) {
+	// todo 如果多处调用 next 可能会导致多实例信息不同步
+	if atomic.CompareAndSwapUint32(&w.first, 0, 1) {
+		return w.getServices()
+	}
 	select {
 	case <-w.ctx.Done():
-		err = w.ctx.Err()
-	case <-w.event:
+		return nil, w.ctx.Err()
+	case e := <-w.event:
+		if e.State == zk.StateDisconnected {
+			return nil, ErrWatcherStopped
+		}
+		if e.Err != nil {
+			return nil, e.Err
+		}
+		return w.getServices()
 	}
-	ss, ok := w.set.services.Load().([]*registry.ServiceInstance)
-	if ok {
-		services = append(services, ss...)
-	}
-	return
 }
 
 func (w *watcher) Stop() error {
 	w.cancel()
-	w.set.lock.Lock()
-	defer w.set.lock.Unlock()
-	delete(w.set.watcher, w)
 	return nil
 }
+
+func (w *watcher) getServices() ([]*registry.ServiceInstance, error) {
+	servicesID, _, err := w.conn.Children(w.prefix)
+	if err != nil {
+		return nil, err
+	}
+	items := make([]*registry.ServiceInstance, 0, len(servicesID))
+	for _, id := range servicesID {
+		servicePath := path.Join(w.prefix, id)
+		b, _, err := w.conn.Get(servicePath)
+		if err != nil {
+			return nil, err
+		}
+		item, err := unmarshal(b)
+		if err != nil {
+			return nil, err
+		}
+
+		// 与 watch 的服务名不同 则跳过
+		if item.Name != w.serviceName {
+			continue
+		}
+
+		items = append(items, item)
+	}
+	return items, nil
+}
diff --git a/errors/errors.go b/errors/errors.go
index f3031616..a18f5512 100644
--- a/errors/errors.go
+++ b/errors/errors.go
@@ -131,20 +131,20 @@ func FromError(err error) *Error {
 		return se
 	}
 	gs, ok := status.FromError(err)
-	if ok {
-		ret := New(
-			httpstatus.FromGRPCCode(gs.Code()),
-			UnknownReason,
-			gs.Message(),
-		)
-		for _, detail := range gs.Details() {
-			switch d := detail.(type) {
-			case *errdetails.ErrorInfo:
-				ret.Reason = d.Reason
-				return ret.WithMetadata(d.Metadata)
-			}
+	if !ok {
+		return New(UnknownCode, UnknownReason, err.Error())
+	}
+	ret := New(
+		httpstatus.FromGRPCCode(gs.Code()),
+		UnknownReason,
+		gs.Message(),
+	)
+	for _, detail := range gs.Details() {
+		switch d := detail.(type) {
+		case *errdetails.ErrorInfo:
+			ret.Reason = d.Reason
+			return ret.WithMetadata(d.Metadata)
 		}
-		return ret
 	}
-	return New(UnknownCode, UnknownReason, err.Error())
+	return ret
 }
diff --git a/internal/endpoint/endpoint.go b/internal/endpoint/endpoint.go
index a4aaf911..a125d924 100644
--- a/internal/endpoint/endpoint.go
+++ b/internal/endpoint/endpoint.go
@@ -21,10 +21,8 @@ func ParseEndpoint(endpoints []string, scheme string, isSecure bool) (string, er
 		if err != nil {
 			return "", err
 		}
-		if u.Scheme == scheme {
-			if IsSecure(u) == isSecure {
-				return u.Host, nil
-			}
+		if u.Scheme == scheme && IsSecure(u) == isSecure {
+			return u.Host, nil
 		}
 	}
 	return "", nil
diff --git a/internal/host/host.go b/internal/host/host.go
index 078660be..d5cc7ffa 100644
--- a/internal/host/host.go
+++ b/internal/host/host.go
@@ -14,9 +14,6 @@ func ExtractHostPort(addr string) (host string, port uint64, err error) {
 		return
 	}
 	port, err = strconv.ParseUint(ports, 10, 16) //nolint:gomnd
-	if err != nil {
-		return
-	}
 	return
 }
 
@@ -40,11 +37,11 @@ func Extract(hostPort string, lis net.Listener) (string, error) {
 		return "", err
 	}
 	if lis != nil {
-		if p, ok := Port(lis); ok {
-			port = strconv.Itoa(p)
-		} else {
+		p, ok := Port(lis)
+		if !ok {
 			return "", fmt.Errorf("failed to extract port: %v", lis.Addr())
 		}
+		port = strconv.Itoa(p)
 	}
 	if len(addr) > 0 && (addr != "0.0.0.0" && addr != "[::]" && addr != "::") {
 		return net.JoinHostPort(addr, port), nil
@@ -61,7 +58,8 @@ func Extract(hostPort string, lis net.Listener) (string, error) {
 		}
 		if iface.Index < lowest || result == nil {
 			lowest = iface.Index
-		} else if result != nil {
+		}
+		if result != nil {
 			continue
 		}
 		addrs, err := iface.Addrs()


Test patch:
diff --git a/contrib/registry/zookeeper/register_test.go b/contrib/registry/zookeeper/register_test.go
index 9293e2b9..8054548d 100644
--- a/contrib/registry/zookeeper/register_test.go
+++ b/contrib/registry/zookeeper/register_test.go
@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/go-kratos/kratos/v2/registry"
+	"github.com/go-zookeeper/zk"
 )
 
 func TestRegistry(t *testing.T) {
@@ -16,11 +17,23 @@ func TestRegistry(t *testing.T) {
 		Endpoints: []string{"http://127.0.0.1:1111"},
 	}
 
-	r, _ := New([]string{"127.0.0.1:2181"}, WithDigestACL("username", "password"))
+	conn, _, err := zk.Connect([]string{"127.0.0.1:2181"}, time.Second*15)
+	if err != nil {
+		t.Fatal(err)
+		return
+	}
+	r := New(conn)
+	err = r.Register(ctx, s)
+	if err != nil {
+		t.Fatal(err)
+		return
+	}
+	time.Sleep(time.Second)
 
 	w, err := r.Watch(ctx, s.Name)
 	if err != nil {
 		t.Fatal(err)
+		return
 	}
 	defer func() {
 		_ = w.Stop()
@@ -29,6 +42,7 @@ func TestRegistry(t *testing.T) {
 		for {
 			res, nextErr := w.Next()
 			if nextErr != nil {
+				t.Errorf("watch next error: %s", nextErr)
 				return
 			}
 			t.Logf("watch: %d", len(res))