// NewRetryWatcher creates a new RetryWatcher.
// It will make sure that watches gets restarted in case of recoverable errors.
// The initialResourceVersion will be given to watch method when first called.
176685
func NewRetryWatcher(initialResourceVersion string, watcherClient cache.Watcher) (*RetryWatcher, error) {
	return newRetryWatcher(initialResourceVersion, watcherClient, 1*time.Second)
}
556963
func NewMockBaseWatcher(ctrl *gomock.Controller) *MockBaseWatcher {
	mock := &MockBaseWatcher{ctrl: ctrl}
	mock.recorder = &MockBaseWatcherMockRecorder{mock}
	return mock
}
580740
func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *string, resyncCh <-chan time.Time, stopCh <-chan struct{}) error {
	start := time.Now()
	eventCount := 0

	// Stopping the watcher should be idempotent and if we return from this function there's no way
	// we're coming back in with the same watch interface.
	defer w.Stop()

loop:
	for {
		select {
		case <-stopCh:
			return errorStopRequested
		case <-resyncCh:
			return errorResyncRequested
		case event, ok := <-w.ResultChan():
			if !ok {
				break loop
			}
			if event.Type == watch.Error {
				return apierrs.FromObject(event.Object)
			}
			if e, a := r.expectedType, reflect.TypeOf(event.Object); e != nil && e != a {
				utilruntime.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", r.name, e, a))
				continue
			}
			meta, err := meta.Accessor(event.Object)
			if err != nil {
				utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
				continue
			}
			newResourceVersion := meta.GetResourceVersion()
			switch event.Type {
			case watch.Added:
				r.store.Add(event.Object)
			case watch.Modified:
				r.store.Update(event.Object)
			case watch.Deleted:
				// TODO: Will any consumers need access to the "last known
				// state", which is passed in event.Object? If so, may need
				// to change this.
				r.store.Delete(event.Object)
			default:
				utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
			}
			*resourceVersion = newResourceVersion
			r.setLastSyncResourceVersion(newResourceVersion)
			eventCount++
		}
	}

	watchDuration := time.Now().Sub(start)
	if watchDuration < 1*time.Second && eventCount == 0 {
		glog.V(4).Infof("%s: Unexpected watch close - watch lasted less than a second and no items received", r.name)
		return errors.New("very short watch")
	}
	glog.V(4).Infof("%s: Watch close - %v total %v items received", r.name, r.expectedType, eventCount)
	return nil
}
139403
func NewWatcher() (*Watcher, error) {
	port, e := syscall.CreateIoCompletionPort(syscall.InvalidHandle, 0, 0, 0)
	if e != nil {
		return nil, os.NewSyscallError("CreateIoCompletionPort", e)
	}
	w := &Watcher{
		port:          port,
		watches:       make(watchMap),
		fsnFlags:      make(map[string]uint32),
		input:         make(chan *input, 1),
		Event:         make(chan *FileEvent, 50),
		internalEvent: make(chan *FileEvent),
		Error:         make(chan error),
		quit:          make(chan chan<- error, 1),
	}
	go w.readEvents()
	go w.purgeEvents()
	return w, nil
}
136949
func (m *Memory) NewWatcher(ctx context.Context, watch backend.Watch) (backend.Watcher, error) {
	if m.EventsOff {
		return nil, trace.BadParameter("events are turned off for this backend")
	}
	return m.buf.NewWatcher(ctx, watch)
}
81971
func (a *AuthServer) NewWatcher(ctx context.Context, watch services.Watch) (services.Watcher, error) {
	return a.GetCache().NewWatcher(ctx, watch)
}
544119
func (a *AbuseRateLimitError) GetRetryAfter() time.Duration {
	if a == nil || a.RetryAfter == nil {
		return 0
	}
	return *a.RetryAfter
}
203017
func NewRetrier(opts Options) Retrier {
	scope := opts.MetricsScope()
	errorTags := struct {
		retryable    map[string]string
		notRetryable map[string]string
	}{
		map[string]string{
			"type": "retryable",
		},
		map[string]string{
			"type": "not-retryable",
		},
	}
	return &retrier{
		initialBackoff: opts.InitialBackoff(),
		backoffFactor:  opts.BackoffFactor(),
		maxBackoff:     opts.MaxBackoff(),
		maxRetries:     opts.MaxRetries(),
		forever:        opts.Forever(),
		jitter:         opts.Jitter(),
		rngFn:          opts.RngFn(),
		sleepFn:        time.Sleep,
		metrics: retrierMetrics{
			success:            scope.Counter("success"),
			successLatency:     scope.Timer("success-latency"),
			errors:             scope.Tagged(errorTags.retryable).Counter("errors"),
			errorsNotRetryable: scope.Tagged(errorTags.notRetryable).Counter("errors"),
			errorsFinal:        scope.Counter("errors-final"),
			errorsLatency:      scope.Timer("errors-latency"),
			retries:            scope.Counter("retries"),
		},
	}
}
801539
func NewOfferStatusWatcher(
	caller base.APICaller, result params.OfferStatusWatchResult,
) watcher.OfferStatusWatcher {
	w := &offerStatusWatcher{
		caller:               caller,
		offerStatusWatcherId: result.OfferStatusWatcherId,
		out:                  make(chan []watcher.OfferStatusChange),
	}
	w.tomb.Go(func() error {
		return w.loop(result.Changes)
	})
	return w
}
1276630
func WatchRepo(userID, repoID int64, watch bool) (err error) {
	return watchRepo(x, userID, repoID, watch)
}
415678
func NewReindexer(client *Client, source string, reindexerFunc ReindexerFunc) *Reindexer {
	return &Reindexer{
		sourceClient:  client,
		sourceIndex:   source,
		reindexerFunc: reindexerFunc,
		statsOnly:     true,
	}
}
895163
func newWaitWorker(config Config, targetVersion int) (worker.Worker, error) {
	watcher, err := config.Facade.WatchModelEnvironVersion(config.ModelTag)
	if err != nil {
		return nil, errors.Trace(err)
	}
	ww := waitWorker{
		watcher:       watcher,
		facade:        config.Facade,
		modelTag:      config.ModelTag,
		gate:          config.GateUnlocker,
		targetVersion: targetVersion,
	}
	if err := catacomb.Invoke(catacomb.Plan{
		Site: &ww.catacomb,
		Init: []worker.Worker{watcher},
		Work: ww.loop,
	}); err != nil {
		return nil, errors.Trace(err)
	}
	return &ww, nil
}
1058038
func newWatch(request *Request, eventChannel *EventChannel) *watch {
	return &watch{
		request:      request,
		eventChannel: eventChannel,
	}
}
507256
func newWatcher(c chan<- EventInfo) watcher {
	r := &readdcw{
		m:   make(map[string]*watched),
		cph: syscall.InvalidHandle,
		c:   c,
	}
	runtime.SetFinalizer(r, func(r *readdcw) {
		if r.cph != syscall.InvalidHandle {
			syscall.CloseHandle(r.cph)
		}
	})
	return r
}
933577
func NewAllWatcher(context facade.Context) (facade.Facade, error) {
	id := context.ID()
	auth := context.Auth()
	resources := context.Resources()

	if !auth.AuthClient() {
		// Note that we don't need to check specific permissions
		// here, as the AllWatcher can only do anything if the
		// watcher resource has already been created, so we can
		// rely on the permission check there to ensure that
		// this facade can't do anything it shouldn't be allowed
		// to.
		//
		// This is useful because the AllWatcher is reused for
		// both the WatchAll (requires model access rights) and
		// the WatchAllModels (requring controller superuser
		// rights) API calls.
		return nil, common.ErrPerm
	}
	watcher, ok := resources.Get(id).(*state.Multiwatcher)
	if !ok {
		return nil, common.ErrUnknownWatcher
	}
	return &SrvAllWatcher{
		watcherCommon: newWatcherCommon(context),
		watcher:       watcher,
	}, nil
}
485715
func NewRetryWaiter(minFailures int, minWait, maxWait time.Duration, jitter Jitter) *RetryWaiter {
	if minFailures < 0 {
		minFailures = defaultMinFailures
	}

	if maxWait <= 0 {
		maxWait = defaultMaxWait
	}

	if minWait <= 0 {
		minWait = 0 * time.Nanosecond
	}

	return &RetryWaiter{
		minFailures: uint(minFailures),
		minWait:     minWait,
		maxWait:     maxWait,
		failures:    0,
		jitter:      jitter,
	}
}
408539
func (l *LiteBackend) NewWatcher(ctx context.Context, watch backend.Watch) (backend.Watcher, error) {
	if l.EventsOff {
		return nil, trace.BadParameter("events are turned off for this backend")
	}
	select {
	case <-l.watchStarted.Done():
	case <-ctx.Done():
		return nil, trace.ConnectionProblem(ctx.Err(), "context is closing")
	}
	return l.buf.NewWatcher(ctx, watch)
}