// WatchCtx is used to wait for either the watch set to trigger or for the
// context to be cancelled. Watch with a timeout channel can be mimicked by
// creating a context with a deadline. WatchCtx should be preferred over Watch.
329599
func (w WatchSet) WatchCtx(ctx context.Context) error {
	if w == nil {
		return nil
	}

	if n := len(w); n <= aFew {
		idx := 0
		chunk := make([]<-chan struct{}, aFew)
		for watchCh := range w {
			chunk[idx] = watchCh
			idx++
		}
		return watchFew(ctx, chunk)
	}

	return w.watchMany(ctx)
}
1298504
func (swi *pluginStatusMap) WatchNameToIdx(subscriber infra.PluginName, pluginChannel chan PluginStatusEvent) {
	swi.mapping.Watch(string(subscriber), func(event idxmap.NamedMappingGenericEvent) {
		pluginChannel <- PluginStatusEvent{
			NamedMappingEvent: event.NamedMappingEvent,
			Value:             swi.castdata(event.Value),
		}
	})
}
598148
func checkWatchSyntax(ctx *cli.Context) {
	if len(ctx.Args()) != 1 {
		cli.ShowCommandHelpAndExit(ctx, "watch", 1) // last argument is exit code
	}
}
1175172
func (r *controller) Wait(pctx context.Context) error {
	if err := r.checkClosed(); err != nil {
		return err
	}

	ctx, cancel := context.WithCancel(pctx)
	defer cancel()

	healthErr := make(chan error, 1)
	go func() {
		ectx, cancel := context.WithCancel(ctx) // cancel event context on first event
		defer cancel()
		if err := r.checkHealth(ectx); err == ErrContainerUnhealthy {
			healthErr <- ErrContainerUnhealthy
			if err := r.Shutdown(ectx); err != nil {
				log.G(ectx).WithError(err).Debug("shutdown failed on unhealthy")
			}
		}
	}()

	waitC, err := r.adapter.wait(ctx)
	if err != nil {
		return err
	}

	if status := <-waitC; status.ExitCode() != 0 {
		exitErr := &exitError{
			code: status.ExitCode(),
		}

		// Set the cause if it is knowable.
		select {
		case e := <-healthErr:
			exitErr.cause = e
		default:
			if status.Err() != nil {
				exitErr.cause = status.Err()
			}
		}

		return exitErr
	}

	return nil
}
102435
func (c *Client) Watch(uri string, ctx context.Context) (<-chan *Response, error) {
	reader, err := c.Do(uri, 0, true)
	if err != nil {
		return nil, err
	}
	respCh := make(chan *Response)
	stop := make(chan struct{})
	go func() {
		select {
		case <-ctx.Done():
		case <-stop:
		}
		reader.Close()
	}()
	go func() {
		defer close(stop)
		defer close(respCh)
		var (
			newone bool = true // if need to create a new response
			resp   *Response
		)
		buf := bufio.NewReader(reader)
		for {
			if newone {
				resp = new(Response)
				resp.finished = false
				newone = false
			}
			line, err := buf.ReadBytes('\n')
			if err != nil {
				if err != io.EOF || len(line) == 0 {
					// got a unexpected(not io.EOF) error, or read none data with io.EOF
					return
				}
			}
			if pureLine := bytes.TrimSpace(line); len(pureLine) == 0 {
				continue // empty line
			} else {
				fields := bytes.SplitN(pureLine, []byte{':'}, 2)
				if len(fields) < 2 {
					continue
				}
				switch key := string(bytes.TrimSpace(fields[0])); key {
				case "id":
					if id, err := strconv.ParseInt(string(bytes.TrimSpace(fields[1])), 10, 64); err == nil {
						resp.Id = id
					}
				case "event":
					resp.Event = string(bytes.TrimSpace(fields[1]))
					if resp.Event == "heartbeat" {
						resp.finished = true
					}
				case "data":
					resp.Data = bytes.TrimSpace(fields[1])
					resp.finished = true
				}
			}
			if resp.finished {
				respCh <- resp
				newone = true
			}
		}
	}()
	return (<-chan *Response)(respCh), nil
}
532675
func (s *Discovery) Watch(stopCh <-chan struct{}) (<-chan discovery.Entries, <-chan error) {
	ch := make(chan discovery.Entries)
	go func() {
		defer close(ch)
		ch <- s.entries
		<-stopCh
	}()
	return ch, nil
}
835383
func (c *FakeTemplates) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(templatesResource, c.ns, opts))

}
1053872
func PostBroadcastTxCtx(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var req apitypes.InsightRawTx
		body, err := ioutil.ReadAll(r.Body)
		r.Body.Close()
		if err != nil {
			writeHTMLBadRequest(w, fmt.Sprintf("Error reading JSON message: %v", err))
			return
		}

		err = json.Unmarshal(body, &req)
		if err != nil {
			writeHTMLBadRequest(w, fmt.Sprintf("Failed to parse request: %v", err))
			return
		}

		// Successful extraction of Body JSON as long as the rawtx is not empty
		// string we should return it.
		if req.Rawtx == "" {
			writeHTMLBadRequest(w, fmt.Sprintf("rawtx cannot be an empty string."))
			return
		}

		ctx := context.WithValue(r.Context(), ctxRawHexTx, req.Rawtx)
		next.ServeHTTP(w, r.WithContext(ctx))
	})
}
782316
func WatchWithRetry(ctx context.Context, getWatch func() (watch.Interface, error)) chan struct {
	*watch.Event
	Error error
} {
	ch := make(chan struct {
		*watch.Event
		Error error
	})
	execute := func() (bool, error) {
		w, err := getWatch()
		if err != nil {
			return false, err
		}
		defer w.Stop()

		for {
			select {
			case event, ok := <-w.ResultChan():
				if ok {
					ch <- struct {
						*watch.Event
						Error error
					}{Event: &event, Error: nil}
				} else {
					return true, nil
				}
			case <-ctx.Done():
				return false, nil
			}
		}
	}
	go func() {
		defer close(ch)
		for {
			retry, err := execute()
			if err != nil {
				ch <- struct {
					*watch.Event
					Error error
				}{Error: err}
			}
			if !retry {
				return
			}
			time.Sleep(time.Second)
		}
	}()
	return ch
}
232642
func (h *allocHealthWatcherHook) watchHealth(ctx context.Context, deadline time.Time, tracker *allochealth.Tracker, done chan<- struct{}) {
	defer close(done)

	// Default to unhealthy for the deadline reached case
	healthy := false

	select {
	case <-ctx.Done():
		// Graceful shutdown
		return

	case <-tracker.AllocStoppedCh():
		// Allocation has stopped so no need to set health
		return

	case <-time.After(deadline.Sub(time.Now())):
		// Time is up! Fallthrough to set unhealthy.
		h.logger.Trace("deadline reached; setting unhealthy", "deadline", deadline)

	case healthy = <-tracker.HealthyCh():
		// Health received. Fallthrough to set it.
	}

	h.logger.Trace("health set", "healthy", healthy)

	// If this is an unhealthy deployment emit events for tasks
	var taskEvents map[string]*structs.TaskEvent
	if !healthy && h.isDeploy {
		taskEvents = tracker.TaskEvents()
	}

	h.healthSetter.SetHealth(healthy, h.isDeploy, taskEvents)
}
1365267
func (c *Context) RefreshTxnCtx(ctx context.Context) error {
	return errors.Trace(c.NewTxn(ctx))
}
450334
func (s *Duration) Watch(watch DurationWatch) {
	s.mutex.Lock()
	defer s.mutex.Unlock()
	s.watches = append(s.watches, watch)
}
229226
func (c *consulClient) Watch(w *Watcher) {
	// Last known state of all KVPairs matching the prefix
	localState := map[string]consulAPI.KVPair{}
	nextIndex := uint64(0)

	qo := consulAPI.QueryOptions{
		WaitTime: time.Second,
	}

	for {
		// Initialize sleep time to a millisecond as we don't
		// want to sleep in between successful watch cycles
		sleepTime := 1 * time.Millisecond

		qo.WaitIndex = nextIndex
		pairs, q, err := c.KV().List(w.prefix, &qo)
		if err != nil {
			sleepTime = 5 * time.Second
			Trace("List of Watch failed", err, logrus.Fields{fieldPrefix: w.prefix, fieldWatcher: w.name})
		}

		if q != nil {
			nextIndex = q.LastIndex
		}

		// timeout while watching for changes, re-schedule
		if qo.WaitIndex != 0 && (q == nil || q.LastIndex == qo.WaitIndex) {
			goto wait
		}

		for _, newPair := range pairs {
			oldPair, ok := localState[newPair.Key]

			// Keys reported for the first time must be new
			if !ok {
				if newPair.CreateIndex != newPair.ModifyIndex {
					log.Debugf("consul: Previously unknown key %s received with CreateIndex(%d) != ModifyIndex(%d)",
						newPair.Key, newPair.CreateIndex, newPair.ModifyIndex)
				}

				queueStart := spanstat.Start()
				w.Events <- KeyValueEvent{
					Typ:   EventTypeCreate,
					Key:   newPair.Key,
					Value: newPair.Value,
				}
				trackEventQueued(newPair.Key, EventTypeCreate, queueStart.End(true).Total())
			} else if oldPair.ModifyIndex != newPair.ModifyIndex {
				queueStart := spanstat.Start()
				w.Events <- KeyValueEvent{
					Typ:   EventTypeModify,
					Key:   newPair.Key,
					Value: newPair.Value,
				}
				trackEventQueued(newPair.Key, EventTypeModify, queueStart.End(true).Total())
			}

			// Everything left on localState will be assumed to
			// have been deleted, therefore remove all keys in
			// localState that still exist in the kvstore
			delete(localState, newPair.Key)
		}

		for k, deletedPair := range localState {
			queueStart := spanstat.Start()
			w.Events <- KeyValueEvent{
				Typ:   EventTypeDelete,
				Key:   deletedPair.Key,
				Value: deletedPair.Value,
			}
			trackEventQueued(deletedPair.Key, EventTypeDelete, queueStart.End(true).Total())
			delete(localState, k)
		}

		for _, newPair := range pairs {
			localState[newPair.Key] = *newPair

		}

		// Initial list operation has been completed, signal this
		if qo.WaitIndex == 0 {
			w.Events <- KeyValueEvent{Typ: EventTypeListDone}
		}

	wait:
		select {
		case <-time.After(sleepTime):
		case <-w.stopWatch:
			close(w.Events)
			w.stopWait.Done()
			return
		}
	}
}
15879
func (lk *etcdLock) waitForLock() {
	// If timeout is not specified, set it to high value
	timeoutIntvl := time.Second * time.Duration(20000)
	if lk.timeout != 0 {
		timeoutIntvl = time.Second * time.Duration(lk.timeout)
	}

	log.Infof("Waiting to acquire lock (%s/%s)", lk.name, lk.myID)

	// Create a timer
	timer := time.NewTimer(timeoutIntvl)
	defer timer.Stop()

	// Wait for changes
	for {
		// wait on watch channel for holder to release the lock
		select {
		case <-timer.C:
			lk.mutex.Lock()
			if lk.timeout != 0 {
				lk.mutex.Unlock()
				log.Infof("Lock timeout on lock %s/%s", lk.name, lk.myID)

				lk.eventChan <- LockEvent{EventType: LockAcquireTimeout}

				log.Infof("Lock acquire timed out. Stopping lock")

				lk.watchCancel()

				// Release the lock
				lk.Release()

				return
			}
			lk.mutex.Unlock()
		case watchResp := <-lk.watchCh:
			if watchResp != nil {
				log.Debugf("Received watch notification(%s/%s): %+v", lk.name, lk.myID, watchResp)

				if watchResp.Action == "expire" || watchResp.Action == "delete" ||
					watchResp.Action == "compareAndDelete" {
					log.Infof("Retrying to acquire lock")
					return
				}
			}
		case <-lk.stopChan:
			log.Infof("Stopping lock")
			lk.watchCancel()
			return
		}
	}
}
1149997
func (w *watchCache) waitUntilFreshAndBlock(resourceVersion uint64, trace *utiltrace.Trace) error {
	startTime := w.clock.Now()
	go func() {
		// Wake us up when the time limit has expired.  The docs
		// promise that time.After (well, NewTimer, which it calls)
		// will wait *at least* the duration given. Since this go
		// routine starts sometime after we record the start time, and
		// it will wake up the loop below sometime after the broadcast,
		// we don't need to worry about waking it up before the time
		// has expired accidentally.
		<-w.clock.After(blockTimeout)
		w.cond.Broadcast()
	}()

	w.RLock()
	if trace != nil {
		trace.Step("watchCache locked acquired")
	}
	for w.resourceVersion < resourceVersion {
		if w.clock.Since(startTime) >= blockTimeout {
			// Timeout with retry after 1 second.
			return errors.NewTimeoutError(fmt.Sprintf("Too large resource version: %v, current: %v", resourceVersion, w.resourceVersion), 1)
		}
		w.cond.Wait()
	}
	if trace != nil {
		trace.Step("watchCache fresh enough")
	}
	return nil
}
717540
func (fw *FirewallSSO) InfoToTemplateCtx(ctx context.Context, info map[string]string, timeout int) map[string]string {
	templateCtx := make(map[string]string)
	for k, v := range info {
		templateCtx[sharedutils.UcFirst(k)] = v
	}

	templateCtx["Timeout"] = strconv.Itoa(timeout)

	return templateCtx
}
591407
func (w *realTimeoutFactory) TimeoutCh() (<-chan time.Time, func() bool) {
	if w.timeout == 0 {
		return neverExitWatch, func() bool { return false }
	}
	t := time.NewTimer(w.timeout)
	return t.C, t.Stop
}