Method watchAndSendKiteEvents

kontrol/_watcher.go:37–149

TODO watchAndSendKiteEvents takes too many arguments. Refactor it.

(
	watcher *Watcher,
	watcherID string,
	disconnect chan bool,
	etcdKey string,
	callback dnode.Function,
	token string,
	hasConstraint bool,
	constraint version.Constraints,
	keyRest string,
)

// TODO watchAndSendKiteEvents takes too many arguments. Refactor it.
func (k *Kontrol) watchAndSendKiteEvents(
	watcher *Watcher,
	watcherID string,
	disconnect chan bool,
	etcdKey string,
	callback dnode.Function,
	token string,
	hasConstraint bool,
	constraint version.Constraints,
	keyRest string,
) {
	var index uint64 = 0
	for {
		select {
		case <-disconnect:
			return
		case resp, ok := <-watcher.recv:
			// Channel is closed. This happens in 3 cases:
			//   1. Remote kite called "cancelWatcher" method and removed the watcher.
			//   2. Remote kite has disconnected and the watcher is removed.
			//   3. Remote kite couldn't consume messages fast enough, buffer
			//      has filled up and etcd cancelled the watcher.
			if !ok {
				// Do not try again if watcher is cancelled.
				k.watchersMutex.Lock()
				if _, ok := k.watchers[watcherID]; !ok {
					k.watchersMutex.Unlock()
					return
				}

				// Do not try again if disconnected.
				select {
				case <-disconnect:
					k.watchersMutex.Unlock()
					return
				default:
				}
				k.watchersMutex.Unlock()

				// If we are here that means we did not consume fast enough and etcd
				// has canceled our watcher. We need to create a new watcher with the same key.
				var err error

				watcher, err = k.storage.Watch(KitesPrefix+etcdKey, index)
				if err != nil {
					k.Kite.Log.Error("Cannot re-watch query: %s", err.Error())
					callback.Call(kite.Response{
						Error: &kite.Error{
							Type:    "watchError",
							Message: err.Error(),
						},
					})
					return
				}

				continue
			}