// parseIOStatWatcher scans the reader for 2 new lines, signifying a new report
502028
func parseIOStatWatcher(r io.Reader, c chan<- map[string]DeviceUtilizationReport, qCh <-chan interface{}) {
	// Custom bufio.Split() function to split tokens by 2 new lines
	atTwoNewLines := func(data []byte, atEOF bool) (int, []byte, error) {
		advance := 0
		var token []byte
		var prev byte

		for _, b := range data {
			// consume the newline by advancing passed it
			if string(b) == "\n" && string(prev) == "\n" {
				advance++
				return advance, token, nil
			}
			token = append(token, b)
			advance++
			prev = b
		}
		return 0, nil, nil
	}

	scanner := bufio.NewScanner(r)
	scanner.Split(atTwoNewLines)
	for scanned := true; scanned; {
		scanc := make(chan struct{})
		go func() {
			defer close(scanc)
			scanned = scanner.Scan()
		}()

		select {
		case <-qCh:
			return
		case <-scanc:
			if scanned {
				out := scanner.Text()
				parseReader := strings.NewReader(out)
				report, err := ParseIOStat(parseReader)
				if err != nil {
					plog.WithError(err).Error("Failed to parse iostat output, exiting")
					return
				} else if len(report) == 0 {
					plog.Warn("Got an empty report from iostat")
					continue
				}
				select {
				case <-qCh:
					return
				case c <- report:
				}
			}
		}
	}
	plog.Warn("parseIOStatWatcher exiting")
	if err := scanner.Err(); err != nil {
		plog.WithError(err).Error("Error reading iostat")
	}
}
35158
func NewWatcher() (*Watcher, error) {
	// Create inotify fd
	fd, errno := unix.InotifyInit1(unix.IN_CLOEXEC)
	if fd == -1 {
		return nil, errno
	}
	// Create epoll
	poller, err := newFdPoller(fd)
	if err != nil {
		unix.Close(fd)
		return nil, err
	}
	w := &Watcher{
		fd:       fd,
		poller:   poller,
		watches:  make(map[string]*watch),
		paths:    make(map[int]string),
		Events:   make(chan Event),
		Errors:   make(chan error),
		done:     make(chan struct{}),
		doneResp: make(chan struct{}),
	}

	go w.readEvents()
	return w, nil
}
1188847
func WatchOneActionReceiverNotifications(tagToActionReceiver func(tag string) (state.ActionReceiver, error), registerFunc func(r facade.Resource) string) func(names.Tag) (params.StringsWatchResult, error) {
	return func(tag names.Tag) (params.StringsWatchResult, error) {
		nothing := params.StringsWatchResult{}
		receiver, err := tagToActionReceiver(tag.String())
		if err != nil {
			return nothing, err
		}
		watch := receiver.WatchActionNotifications()

		if changes, ok := <-watch.Changes(); ok {
			return params.StringsWatchResult{
				StringsWatcherId: registerFunc(watch),
				Changes:          changes,
			}, nil
		}
		return nothing, watcher.EnsureErr(watch)
	}
}
291769
func NewAgentEntityWatcher(st state.EntityFinder, resources facade.Resources, getCanWatch GetAuthFunc) *AgentEntityWatcher {
	return &AgentEntityWatcher{
		st:          st,
		resources:   resources,
		getCanWatch: getCanWatch,
	}
}
220535
func (p *TextParser) parseError(msg string) {
	p.err = ParseError{
		Line: p.lineCount,
		Msg:  msg,
	}
}
262278
func newStreamWatcher(r io.ReadCloser) watchInterface {
	sw := &streamWatcher{
		r:       r,
		decoder: json.NewDecoder(r),
		result:  make(chan Event),
	}
	go sw.receive()
	return sw
}
198458
func checksWatch(params map[string]interface{}) (WatcherFunc, error) {
	stale := false
	if err := assignValueBool(params, "stale", &stale); err != nil {
		return nil, err
	}

	var service, state string
	if err := assignValue(params, "service", &service); err != nil {
		return nil, err
	}
	if err := assignValue(params, "state", &state); err != nil {
		return nil, err
	}
	if service != "" && state != "" {
		return nil, fmt.Errorf("Cannot specify service and state")
	}
	if service == "" && state == "" {
		state = "any"
	}

	fn := func(p *Plan) (BlockingParamVal, interface{}, error) {
		health := p.client.Health()
		opts := makeQueryOptionsWithContext(p, stale)
		defer p.cancelFunc()
		var checks []*consulapi.HealthCheck
		var meta *consulapi.QueryMeta
		var err error
		if state != "" {
			checks, meta, err = health.State(state, &opts)
		} else {
			checks, meta, err = health.Checks(service, &opts)
		}
		if err != nil {
			return nil, nil, err
		}
		return WaitIndexVal(meta.LastIndex), checks, err
	}
	return fn, nil
}
688053
func NewEventWatcher() (FileWatcher, error) {
	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		return nil, err
	}
	return &fsNotifyWatcher{watcher}, nil
}
1186577
func newConfigWatcher(
	keys []string, cache *hashCache, hub *pubsub.SimpleHub, topic string, res *Resident,
) *ConfigWatcher {
	sort.Strings(keys)

	w := &ConfigWatcher{
		notifyWatcherBase: newNotifyWatcherBase(),

		keys: keys,
		hash: cache.getHash(keys),
	}

	deregister := res.registerWorker(w)
	unsub := hub.Subscribe(topic, w.configChanged)
	w.tomb.Go(func() error {
		<-w.tomb.Dying()
		unsub()
		deregister()
		return nil
	})

	return w
}
37029
func (w *Watcher) getOrCreateWatcher(dID string) (*deploymentWatcher, error) {
	w.l.Lock()
	defer w.l.Unlock()

	// Not enabled so no-op
	if !w.enabled {
		return nil, notEnabled
	}

	watcher, ok := w.watchers[dID]
	if ok {
		return watcher, nil
	}

	return w.forceAdd(dID)
}
1200128
func newWatcher(c chan<- EventInfo) watcher {
	i := &inotify{
		m:      make(map[int32]*watched),
		fd:     invalidDescriptor,
		pipefd: []int{invalidDescriptor, invalidDescriptor},
		epfd:   invalidDescriptor,
		epes:   make([]unix.EpollEvent, 0),
		c:      c,
	}
	runtime.SetFinalizer(i, func(i *inotify) {
		i.epollclose()
		if i.fd != invalidDescriptor {
			unix.Close(int(i.fd))
		}
	})
	return i
}
1097275
func (p *parser) parseIDSelector() (Selector, error) {
	if p.i >= len(p.s) {
		return nil, fmt.Errorf("expected id selector (#id), found EOF instead")
	}
	if p.s[p.i] != '#' {
		return nil, fmt.Errorf("expected id selector (#id), found '%c' instead", p.s[p.i])
	}

	p.i++
	id, err := p.parseName()
	if err != nil {
		return nil, err
	}

	return attributeEqualsSelector("id", id), nil
}
146880
func NewWatcher() (*Watcher, error) {
	kq, err := kqueue()
	if err != nil {
		return nil, err
	}

	w := &Watcher{
		kq:              kq,
		watches:         make(map[string]int),
		dirFlags:        make(map[string]uint32),
		paths:           make(map[int]pathInfo),
		fileExists:      make(map[string]bool),
		externalWatches: make(map[string]bool),
		Events:          make(chan Event),
		Errors:          make(chan error),
		done:            make(chan struct{}),
	}

	go w.readEvents()
	return w, nil
}
460259
func (m *NodeManager) CloseWatcher(i int) {
	m.mu.Lock()
	defer m.mu.Unlock()

	delete(m.watchers, i)
}
519376
func NewWatcher() *Watcher {
	w := &Watcher{
		pins:         make(map[uintptr]Pin),
		fds:          fdHeap{},
		cmdChan:      make(chan watcherCmd, watcherCmdChanLen),
		Notification: make(chan WatcherNotification, notificationLen),
	}
	heap.Init(&w.fds)
	go w.watch()
	return w
}
1135001
func (api *API) WatchForProxyConfigAndAPIHostPortChanges() (watcher.NotifyWatcher, error) {
	var results params.NotifyWatchResults
	args := params.Entities{
		Entities: []params.Entity{{Tag: api.tag.String()}},
	}
	err := api.facade.FacadeCall("WatchForProxyConfigAndAPIHostPortChanges", args, &results)
	if err != nil {
		return nil, err
	}
	if len(results.Results) != 1 {
		return nil, errors.Errorf("expected 1 result, got %d", len(results.Results))
	}
	result := results.Results[0]
	if result.Error != nil {
		return nil, result.Error
	}

	return newNotifyWatcher(api.facade.RawAPICaller(), result), nil
}
1142600
func NewRelationStatusWatcher(
	caller base.APICaller, result params.RelationLifeSuspendedStatusWatchResult,
) watcher.RelationStatusWatcher {
	w := &relationStatusWatcher{
		caller:                  caller,
		relationStatusWatcherId: result.RelationStatusWatcherId,
		out:                     make(chan []watcher.RelationStatusChange),
	}
	w.tomb.Go(func() error {
		return w.loop(result.Changes)
	})
	return w
}