// NewWatcher creates a new watcher with the given retry function and
// interval generator.
161664
func NewWatcher(retry Retry, backoff backoff.Backoff) Watcher {
	return newWatcherWithClock(retry, backoff, glock.NewRealClock())
}
508788
func NewWorker(api Facade, env environs.Environ, credentialAPI common.CredentialAPI) (worker.Worker, error) {
	envNetworking, _ := environs.SupportsNetworking(env)
	w, err := watcher.NewNotifyWorker(watcher.NotifyConfig{
		Handler: &Undertaker{
			API:         api,
			Releaser:    envNetworking,
			CallContext: common.NewCloudCallContext(credentialAPI, nil),
		},
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return w, nil
}
1100889
func New() (FileWatcher, error) {
	if watcher, err := NewEventWatcher(); err == nil {
		return watcher, nil
	}
	return NewPollingWatcher(), nil
}
1092443
func NewAgentConfigWatcher(client *api.Client, proxyID string,
	logger *log.Logger) (*AgentConfigWatcher, error) {
	w := &AgentConfigWatcher{
		client:  client,
		proxyID: proxyID,
		logger:  logger,
		ch:      make(chan *Config),
	}

	// Setup watch plan for config
	plan, err := watch.Parse(map[string]interface{}{
		"type":       "agent_service",
		"service_id": w.proxyID,
	})
	if err != nil {
		return nil, err
	}
	w.plan = plan
	w.plan.HybridHandler = w.handler
	go w.plan.RunWithClientAndLogger(w.client, w.logger)
	return w, nil
}
796831
func NewWatcher(sessionStartTime time.Time) *Watcher {
	return &Watcher{
		sessionStartTime: sessionStartTime,
		errorChan:        make(chan *probe.Error),
		eventInfoChan:    make(chan EventInfo),
		o:                []*watchObject{},
	}
}
737011
func NewCacherListerWatcher(storage storage.Interface, resourcePrefix string, newListFunc func() runtime.Object) cache.ListerWatcher {
	return &cacherListerWatcher{
		storage:        storage,
		resourcePrefix: resourcePrefix,
		newListFunc:    newListFunc,
	}
}
1218917
func NewBytesWatcherWrapper(pb keyval.BytesWatcher, decrypter ArbitraryDecrypter, decryptFunc DecryptFunc) *BytesWatcherWrapper {
	return &BytesWatcherWrapper{
		BytesWatcher: pb,
		decryptData: decryptData{
			decryptFunc: decryptFunc,
			decrypter:   decrypter,
		},
	}
}
980945
func (db *ProtoWrapper) NewWatcher(prefix string) keyval.ProtoWatcher {
	return &protoWatcher{db.broker.NewWatcher(prefix), db.serializer}
}
1130379
func NewLogWatcher() *LogWatcher {
	return &LogWatcher{
		Msg:          make(chan *Message, logWatcherBufferSize),
		Err:          make(chan error, 1),
		producerGone: make(chan struct{}),
		consumerGone: make(chan struct{}),
	}
}
91315
func newRetryTimer(unit time.Duration, cap time.Duration, doneCh chan struct{}) <-chan int {
	return newRetryTimerWithJitter(unit, cap, MaxJitter, doneCh)
}
981285
func (eventBroadcaster *eventBroadcasterImpl) StartEventWatcher(eventHandler func(*v1.Event)) watch.Interface {
	watcher := eventBroadcaster.Watch()
	go func() {
		defer utilruntime.HandleCrash()
		for watchEvent := range watcher.ResultChan() {
			event, ok := watchEvent.Object.(*v1.Event)
			if !ok {
				// This is all local, so there's no reason this should
				// ever happen.
				continue
			}
			eventHandler(event)
		}
	}()
	return watcher
}
556963
func NewMockBaseWatcher(ctrl *gomock.Controller) *MockBaseWatcher {
	mock := &MockBaseWatcher{ctrl: ctrl}
	mock.recorder = &MockBaseWatcherMockRecorder{mock}
	return mock
}
135257
func NewMultiNotifyWatcher(w ...NotifyWatcher) *MultiNotifyWatcher {
	m := &MultiNotifyWatcher{
		watchers: w,
		changes:  make(chan struct{}),
	}
	var wg sync.WaitGroup
	wg.Add(len(w))
	staging := make(chan struct{})
	for _, w := range w {
		// Consume the first event of each watcher.
		<-w.Changes()
		go func(wCopy NotifyWatcher) {
			defer wg.Done()
			wCopy.Wait()
		}(w)
		// Copy events from the watcher to the staging channel.
		go copyEvents(staging, w.Changes(), &m.tomb)
	}
	m.tomb.Go(func() error {
		m.loop(staging)
		wg.Wait()
		return nil
	})
	return m
}
705731
func (c Client) newRetryTimer(maxRetry int, unit time.Duration, cap time.Duration, jitter float64, doneCh chan struct{}) <-chan int {
	attemptCh := make(chan int)

	// computes the exponential backoff duration according to
	// https://www.awsarchitectureblog.com/2015/03/backoff.html
	exponentialBackoffWait := func(attempt int) time.Duration {
		// normalize jitter to the range [0, 1.0]
		if jitter < NoJitter {
			jitter = NoJitter
		}
		if jitter > MaxJitter {
			jitter = MaxJitter
		}

		//sleep = random_between(0, min(cap, base * 2 ** attempt))
		sleep := unit * time.Duration(1<<uint(attempt))
		if sleep > cap {
			sleep = cap
		}
		if jitter != NoJitter {
			sleep -= time.Duration(c.random.Float64() * float64(sleep) * jitter)
		}
		return sleep
	}

	go func() {
		defer close(attemptCh)
		for i := 0; i < maxRetry; i++ {
			select {
			// Attempts start from 1.
			case attemptCh <- i + 1:
			case <-doneCh:
				// Stop the routine.
				return
			}
			time.Sleep(exponentialBackoffWait(i))
		}
	}()
	return attemptCh
}
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
}
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
}
1347324
func NewMultiwatcher(all *storeManager) *Multiwatcher {
	// Note that we want to be clear about the defaults. So we set zero
	// values explicitly.
	//  used:    false means that the watcher has not been used yet
	//  revno:   0 means that *all* transactions prior to the first
	//           Next() call will be reflected in the deltas.
	//  stopped: false means that the watcher immediately starts off
	//           handling changes.
	return &Multiwatcher{
		all:     all,
		used:    false,
		revno:   0,
		stopped: false,
	}
}