Method processWatchLoop

server/sync_watcher.go:138–206

processWatchLoop handles events from the watch on the process list. When this method detects a change, spawns new goroutines for sync event handling. This method blocks until the events channel is closed by the caller or an error event is given from the channel.

(events <-chan *gohan_sync.Event)

// This method blocks until the events channel is closed by the caller or
// an error event is given from the channel.
func (watcher *SyncWatcher) processWatchLoop(events <-chan *gohan_sync.Event) error {
	processList := []string{}

	previousCancel := func() {}
	previousDone := make(chan struct{})
	close(previousDone)
	defer func() {
		previousCancel()
		<-previousDone
	}()

	for event := range events {
		previousCancel()
		<-previousDone

		if event.Err != nil {
			return event.Err
		}

		log.Debug("cluster change detected: %s process %s", event.Action, event.Key)

		// modify gohan process list
		pos := -1
		for p, v := range processList {
			if v == event.Key {
				pos = p
			}
		}
		switch event.Action {
		case "delete":
			// remove detected process from list
			if pos > -1 {
				processList = append((processList)[:pos], (processList)[pos+1:]...)
			}
		default:
			// add detected process from list
			if pos == -1 {
				processList = append(processList, event.Key)
				sort.Sort(sort.StringSlice(processList))
			}
		}

		myPosition := -1
		myValue := processPathPrefix + "/" + watcher.sync.GetProcessID()
		for p, v := range processList {
			if v == myValue {
				myPosition = p
				break
			}
		}

		if myPosition >= 0 && len(processList) > 0 {
			log.Debug("Current cluster consists of following processes: %s, my poistion: %d", processList, myPosition)

			var cctx context.Context
			cctx, previousCancel = context.WithCancel(context.Background())
			previousDone = make(chan struct{})