// NewPartitionWatcher creates a watcher that allows to start/stop consuming of messaging published to given topics, offset and partition
167822
func (p *Plugin) NewPartitionWatcher(name string) messaging.ProtoPartitionWatcher {
	return p.NewProtoManualConnection(name)
}
1093158
func (c *consulClient) ListAndWatch(name, prefix string, chanSize int) *Watcher {
	w := newWatcher(name, prefix, chanSize)

	log.WithField(fieldWatcher, w).Debug("Starting watcher...")

	go c.Watch(w)

	return w
}
463852
func (conn *ProtoManualConnection) WatchPartition(msgClb func(messaging.ProtoMessage), topic string, partition int32, offset int64) error {
	return conn.ConsumePartition(msgClb, topic, partition, offset)
}
796831
func NewWatcher(sessionStartTime time.Time) *Watcher {
	return &Watcher{
		sessionStartTime: sessionStartTime,
		errorChan:        make(chan *probe.Error),
		eventInfoChan:    make(chan EventInfo),
		o:                []*watchObject{},
	}
}
185691
func NewPartition(r io.Reader) (*Partition, error) {
	scanner := bufio.NewScanner(r)
	words := make(map[string]int)
	classes := make(map[int][]string)

	i := 0
	for scanner.Scan() {
		line := scanner.Text()

		fields := strings.Fields(line)
		if len(fields) != 2 {
			return nil, fmt.Errorf("[line: %d] expected 2 fields, instead got: %v", i+1, len(fields))
		}

		w := fields[0]
		c, err := strconv.Atoi(fields[1])
		if err != nil {
			return nil, fmt.Errorf("[line: %d] error parsing integer %#v: %v", i+1, fields[1], err)
		}

		words[w] = c
		classes[c] = append(classes[c], w)
		i++
	}

	err := scanner.Err()
	if err != nil {
		return nil, fmt.Errorf("[line: %d] scanner error: %v", i+1, err)
	}

	return &Partition{
		words:   words,
		classes: classes,
	}, nil
}
212209
func NewSyncConsumer(brokers []string, topic string, group string, offset int64) (*ConsumerGroup, error) {
	return newConsumer(brokers, topic, group, offset, true)
}
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
}
1221022
func NewPartition(pool Pool, buffers ...Buffer) Buffer {
	return &partition{
		Pool: pool,
		List: buffers,
	}
}
960806
func (manager *configManager) WatchCluster(name string, cb func(config *ClusterConfig) error) error {
	manager.Lock()
	defer manager.Unlock()

	if _, present := manager.cbCluster[name]; present {
		return fmt.Errorf("%s already present", name)
	}
	manager.cbCluster[name] = cb
	return nil
}
1145226
func (sb *storageBackend) WatchFilesystemAttachment(host names.Tag, f names.FilesystemTag) NotifyWatcher {
	id := filesystemAttachmentId(host.Id(), f.Id())
	return newEntityWatcher(sb.mb, filesystemAttachmentsC, sb.mb.docID(id))
}
394920
func (k *kubernetesClient) WatchUnits(appName string) (watcher.NotifyWatcher, error) {
	selector := applicationSelector(appName)
	logger.Debugf("selecting units %q to watch", selector)
	w, err := k.client().CoreV1().Pods(k.namespace).Watch(v1.ListOptions{
		LabelSelector:        selector,
		Watch:                true,
		IncludeUninitialized: true,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return k.newWatcher(w, appName, k.clock)
}
262278
func newStreamWatcher(r io.ReadCloser) watchInterface {
	sw := &streamWatcher{
		r:       r,
		decoder: json.NewDecoder(r),
		result:  make(chan Event),
	}
	go sw.receive()
	return sw
}
503337
func (c *CircularBuffer) NewWatcher(ctx context.Context, watch Watch) (Watcher, error) {
	c.Lock()
	defer c.Unlock()

	select {
	case <-c.ctx.Done():
		return nil, trace.BadParameter("buffer is closed")
	default:
	}

	if watch.QueueSize == 0 {
		watch.QueueSize = len(c.events)
	}

	if len(watch.Prefixes) == 0 {
		// if watcher has no prefixes, assume it will match anything
		// starting from the separator (what includes all keys in backend invariant, see Keys function)
		watch.Prefixes = append(watch.Prefixes, []byte{Separator})
	} else {
		// if watcher's prefixes are redundant, keep only shorter prefixes
		// to avoid double fan out
		watch.Prefixes = removeRedundantPrefixes(watch.Prefixes)
	}

	closeCtx, cancel := context.WithCancel(ctx)
	w := &BufferWatcher{
		Watch:    watch,
		eventsC:  make(chan Event, watch.QueueSize),
		ctx:      closeCtx,
		cancel:   cancel,
		capacity: watch.QueueSize,
	}
	c.Debugf("Add %v.", w)
	select {
	case w.eventsC <- Event{Type: OpInit}:
	case <-c.ctx.Done():
		return nil, trace.BadParameter("buffer is closed")
	default:
		c.Warningf("Closing %v, buffer overflow.", w)
		w.Close()
		return nil, trace.BadParameter("buffer overflow")
	}
	c.watchers.add(w)
	return w, nil
}
62445
func NewUnitsWatcher(st state.EntityFinder, resources facade.Resources, getCanWatch GetAuthFunc) *UnitsWatcher {
	return &UnitsWatcher{
		st:          st,
		resources:   resources,
		getCanWatch: getCanWatch,
	}
}
1115266
func (st *State) watchEnqueuedActionsFilteredBy(receivers ...ActionReceiver) StringsWatcher {
	return newCollectionWatcher(st, colWCfg{
		col:    actionNotificationsC,
		filter: makeIdFilter(st, actionMarker, receivers...),
		idconv: actionNotificationIdToActionId,
	})
}
909871
func NewRandomPartitioner(topic string) Partitioner {
	p := new(randomPartitioner)
	p.generator = rand.New(rand.NewSource(time.Now().UTC().UnixNano()))
	return p
}
100113
func NewAllWatcher(caller base.APICaller, id *string) *AllWatcher {
	return newAllWatcher("AllWatcher", caller, id)
}