watch branch

gossip.go

@@ -418,9 +418,9 @@ func (m *gossipRegistry) ListServices() ([]*registry.Service, error) {
 	return services, nil
 }
 
-func (m *gossipRegistry) Watch() (registry.Watcher, error) {
+func (m *gossipRegistry) Watch(opts ...registry.WatchOption) (registry.Watcher, error) {
 	n, e := m.subscribe()
-	return newGossipWatcher(n, e)
+	return newGossipWatcher(n, e, opts...)
 }
 
 func (m *gossipRegistry) String() string {

watch.go

@@ -7,11 +7,17 @@ import (
 )
 
 type gossipWatcher struct {
+	wo   registry.WatchOptions
 	next chan *registry.Result
 	stop chan bool
 }
 
-func newGossipWatcher(ch chan *registry.Result, exit chan bool) (registry.Watcher, error) {
+func newGossipWatcher(ch chan *registry.Result, exit chan bool, opts ...registry.WatchOption) (registry.Watcher, error) {
+	var wo registry.WatchOptions
+	for _, o := range opts {
+		o(&wo)
+	}
+
 	stop := make(chan bool)
 
 	go func() {
@@ -20,22 +26,29 @@ func newGossipWatcher(ch chan *registry.Result, exit chan bool) (registry.Watche
 	}()
 
 	return &gossipWatcher{
+		wo:   wo,
 		next: ch,
 		stop: stop,
 	}, nil
 }
 
 func (m *gossipWatcher) Next() (*registry.Result, error) {
+	for {
 		select {
 		case r, ok := <-m.next:
 			if !ok {
 				return nil, errors.New("result chan closed")
 			}
+			// check watch options
+			if len(m.wo.Service) > 0 && r.Service.Name != m.wo.Service {
+				continue
+			}
 			return r, nil
 		case <-m.stop:
 			return nil, errors.New("watcher stopped")
 		}
 	}
+}
 
 func (m *gossipWatcher) Stop() {
 	select {