Method Subscribe

engine/etcdng/v3/etcd.go:625–679

Subscribe watches etcd changes and generates structured events telling vulcand to add or delete frontends, hosts etc. It is a blocking function.

(changes chan interface{}, afterIdx uint64, cancelC chan struct{})

// Subscribe watches etcd changes and generates structured events telling vulcand to add or delete frontends, hosts etc.
// It is a blocking function.
func (n *ng) Subscribe(changes chan interface{}, afterIdx uint64, cancelC chan struct{}) error {
	watcher := etcd.NewWatcher(n.client)
	defer watcher.Close()

	var watchChan etcd.WatchChan
	ready := make(chan struct{})
	go func() {
		watchChan = watcher.Watch(etcd.WithRequireLeader(n.context), n.etcdKey,
			etcd.WithRev(int64(afterIdx)), etcd.WithPrefix())
		close(ready)
	}()
	select {
	case <-ready:
		log.Infof("begin watching: etcd revision %d", afterIdx)
	case <-time.After(time.Second * 10):
		return errors.New("timed out while waiting for watcher.Watch() to start")
	}

	for response := range watchChan {
		if response.Canceled {
			log.Warn("etcd watcher cancelled")

			if err := response.Err(); err != nil {
				log.Errorf("etcd watcher cancelled with error: %v", err)
				return err
			}

			return errors.New("etcd watcher failed without error message (canceled)")
		}

		if err := response.Err(); err != nil {
			log.Errorf("etcd watcher received error: %v", err)
			return err
		}

		for _, event := range response.Events {
			log.WithFields(eventToFields(event)).Infof("%s: %s", event.Type, event.Kv.Key)
			change, err := n.parseChange(event)
			if err != nil {
				log.WithFields(eventToFields(event)).Warningf("ignoring event; error: %s", err)
				continue
			}
			if change != nil {
				log.Infof("%v", change)
				select {
				case changes <- change:
				case <-cancelC:
					return nil
				}
			}
		}
	}

	return errors.New("etcd watcher channel closed without graceful stop")
}

type MatcherFn func(*etcd.Event) (interface{}, error)