Method Run

server/sync_watcher.go:89–132

Run starts the main loop of the watcher. This method blocks until the ctx is canceled by the caller

(ctx context.Context)

// Run starts the main loop of the watcher.
// This method blocks until the ctx is canceled by the caller
func (watcher *SyncWatcher) Run(ctx context.Context) error {
	for {
		err := func() error {
			// register self process to the cluster
			lockKey := processPathPrefix + "/" + watcher.sync.GetProcessID()
			lost, err := watcher.sync.Lock(lockKey, true)
			if err != nil {
				return err
			}
			defer watcher.sync.Unlock(lockKey)

			watchCtx, watchCancel := context.WithCancel(ctx)
			defer watchCancel()
			events := watcher.sync.WatchContext(watchCtx, processPathPrefix, int64(gohan_sync.RevisionCurrent))
			watchErr := make(chan error, 1)
			go func() {
				watchErr <- watcher.processWatchLoop(events)
			}()

			select {
			case err := <-watchErr:
				return err
			case <-ctx.Done():
				<-watchErr
				return nil
			case <-lost:
				watchCancel()
				<-watchErr
				return fmt.Errorf("lock is lost")
			}
		}()

		if err != nil {
			log.Error("process watch intrupted: %s", err)
		}

		select {
		case <-time.After(watcher.backoff):
			continue
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

// processWatchLoop handles events from the watch on the process list.
// When this method detects a change, spawns new goroutines for sync event handling.