// subscribeWatcher subscribes for data change and data resync events.
// Events are delivered to the consumer via the selected channels.
// ETCD watcher adapter is used to perform the registration behind the scenes.
505065
func (p *ExamplePlugin) subscribeWatcher() (err error) {
	prefix := etcdKeyPrefix(p.ServiceLabel.GetAgentLabel())
	p.Log.Infof("Prefix: %v", prefix)

	p.watchDataReg, err = p.Watcher.Watch("ExamplePlugin",
		p.changeChannel, p.resyncChannel, prefix)
	if err != nil {
		return err
	}

	p.Log.Info("KeyValProtoWatcher subscribed")

	return nil
}
775878
func (ep *EtcdClient) WatchService(name string, eventCh chan WatchServiceEvent, stopCh chan bool) error {
	keyName := "/contiv.io/service/" + name + "/"

	// Create channels
	watchCh := make(chan *client.Response, 1)

	// Create watch context
	watchCtx, watchCancel := context.WithCancel(context.Background())

	// Start the watch thread
	go func() {
		// Get current state and etcd index to watch
		watchIndex, err := ep.initServiceState(keyName, eventCh)
		if err != nil {
			log.Fatalf("Unable to watch service key: %s - %v", keyName,
				err)
		}

		log.Infof("Watching for service: %s at index %v", keyName, watchIndex)
		// Start the watch
		watcher := ep.kapi.Watcher(keyName, &client.WatcherOptions{AfterIndex: watchIndex, Recursive: true})
		if watcher == nil {
			log.Errorf("Error watching service %s. Etcd returned invalid watcher", keyName)

			// Emit the event
			eventCh <- WatchServiceEvent{EventType: WatchServiceEventError}
		}

		// Keep getting next event
		for {
			// Block till next watch event
			etcdRsp, err := watcher.Next(watchCtx)
			if err != nil && err.Error() == client.ErrClusterUnavailable.Error() {
				log.Infof("Stopping watch on key %s", keyName)
				return
			} else if err != nil {
				log.Errorf("Error %v during watch. Watch thread exiting", err)
				return
			}

			// Send it to watch channel
			watchCh <- etcdRsp
		}
	}()

	// handle messages from watch service
	go func() {
		var srvMap = make(map[string]ServiceInfo)
		for {
			select {
			case watchResp := <-watchCh:
				var srvInfo ServiceInfo

				log.Debugf("Received event {%#v}\n Node: {%#v}\n PrevNade: {%#v}", watchResp, watchResp.Node, watchResp.PrevNode)

				// derive service info from key
				srvKey := strings.TrimPrefix(watchResp.Node.Key, "/contiv.io/service/")

				// We ignore all events except Set/Delete/Expire
				// Note that Set event doesnt exactly mean new service end point.
				// If a service restarts and re-registers before it expired, we'll
				// receive set again. receivers need to handle this case
				if _, ok := srvMap[srvKey]; !ok && watchResp.Action == "set" {
					// Parse JSON response
					err := json.Unmarshal([]byte(watchResp.Node.Value), &srvInfo)
					if err != nil {
						log.Errorf("Error parsing object %s, Err %v", watchResp.Node.Value, err)
						break
					}

					log.Infof("Sending service add event: %+v", srvInfo)
					// Send Add event
					eventCh <- WatchServiceEvent{
						EventType:   WatchServiceEventAdd,
						ServiceInfo: srvInfo,
					}

					// save it in cache
					srvMap[srvKey] = srvInfo
				} else if (watchResp.Action == "delete") || (watchResp.Action == "expire") {
					// Parse JSON response
					err := json.Unmarshal([]byte(watchResp.PrevNode.Value), &srvInfo)
					if err != nil {
						log.Errorf("Error parsing object %s, Err %v", watchResp.Node.Value, err)
						break
					}

					log.Infof("Sending service del event: %+v", srvInfo)

					// Send Delete event
					eventCh <- WatchServiceEvent{
						EventType:   WatchServiceEventDel,
						ServiceInfo: srvInfo,
					}

					// remove it from cache
					delete(srvMap, srvKey)
				}
			case stopReq := <-stopCh:
				if stopReq {
					// Stop watch and return
					log.Infof("Stopping watch on %s", keyName)
					watchCancel()
					return
				}
			}
		}
	}()

	return nil
}
738696
func (ccb *ccBalancerWrapper) watcher() {
	for {
		select {
		case t := <-ccb.stateChangeQueue.get():
			ccb.stateChangeQueue.load()
			select {
			case <-ccb.done:
				ccb.balancer.Close()
				return
			default:
			}
			if ub, ok := ccb.balancer.(balancer.V2Balancer); ok {
				ub.UpdateSubConnState(t.sc, balancer.SubConnState{ConnectivityState: t.state})
			} else {
				ccb.balancer.HandleSubConnStateChange(t.sc, t.state)
			}
		case s := <-ccb.resolverUpdateCh:
			select {
			case <-ccb.done:
				ccb.balancer.Close()
				return
			default:
			}
			if ub, ok := ccb.balancer.(balancer.V2Balancer); ok {
				ub.UpdateResolverState(*s)
			} else {
				ccb.balancer.HandleResolvedAddrs(s.Addresses, nil)
			}
		case <-ccb.done:
		}

		select {
		case <-ccb.done:
			ccb.balancer.Close()
			ccb.mu.Lock()
			scs := ccb.subConns
			ccb.subConns = nil
			ccb.mu.Unlock()
			for acbw := range scs {
				ccb.cc.removeAddrConn(acbw.getAddrConn(), errConnDrain)
			}
			return
		default:
		}
	}
}
262278
func newStreamWatcher(r io.ReadCloser) watchInterface {
	sw := &streamWatcher{
		r:       r,
		decoder: json.NewDecoder(r),
		result:  make(chan Event),
	}
	go sw.receive()
	return sw
}
258493
func (c *EncodedConn) Subscribe(subject string, cb Handler) (*Subscription, error) {
	return c.subscribe(subject, _EMPTY_, cb)
}
1117251
func (e *EventsService) NewWatcher(ctx context.Context, watch services.Watch) (services.Watcher, error) {
	if len(watch.Kinds) == 0 {
		return nil, trace.BadParameter("global watches are not supported yet")
	}
	var parsers []resourceParser
	var prefixes [][]byte
	for _, kind := range watch.Kinds {
		if kind.Name != "" && kind.Kind != services.KindNamespace {
			return nil, trace.BadParameter("watch with Name is only supported for Namespace resource")
		}
		var parser resourceParser
		switch kind.Kind {
		case services.KindCertAuthority:
			parser = newCertAuthorityParser(kind.LoadSecrets)
		case services.KindToken:
			parser = newProvisionTokenParser()
		case services.KindStaticTokens:
			parser = newStaticTokensParser()
		case services.KindClusterConfig:
			parser = newClusterConfigParser()
		case services.KindClusterName:
			parser = newClusterNameParser()
		case services.KindNamespace:
			parser = newNamespaceParser(kind.Name)
		case services.KindRole:
			parser = newRoleParser()
		case services.KindUser:
			parser = newUserParser()
		case services.KindNode:
			parser = newNodeParser()
		case services.KindProxy:
			parser = newProxyParser()
		case services.KindAuthServer:
			parser = newAuthServerParser()
		case services.KindTunnelConnection:
			parser = newTunnelConnectionParser()
		case services.KindReverseTunnel:
			parser = newReverseTunnelParser()
		default:
			return nil, trace.BadParameter("watcher on object kind %v is not supported", kind)
		}
		prefixes = append(prefixes, parser.prefix())
		parsers = append(parsers, parser)
	}
	// sort so that longer prefixes get first
	sort.Slice(parsers, func(i, j int) bool { return len(parsers[i].prefix()) > len(parsers[j].prefix()) })
	w, err := e.backend.NewWatcher(ctx, backend.Watch{
		Name:            watch.Name,
		Prefixes:        prefixes,
		QueueSize:       watch.QueueSize,
		MetricComponent: watch.MetricComponent,
	})
	if err != nil {
		return nil, trace.Wrap(err)
	}
	return newWatcher(w, e.Entry, parsers), nil
}
835075
func NewWatcher() (*RWatcher, error) {
	fsWatch, err := fsnotify.NewWatcher()
	if err != nil {
		return nil, err
	}

	m := &RWatcher{}
	m.fsnotify = fsWatch
	m.Events = make(chan fsnotify.Event)
	m.Errors = make(chan error)
	m.done = make(chan struct{})

	go m.start()

	return m, nil
}
1296214
func (server *RateServer) SubscribeExchanges(hello *dcrrates.ExchangeSubscription, stream dcrrates.DCRRates_SubscribeExchangesServer) (err error) {
	return server.ReallySubscribeExchanges(hello, stream)
}
648461
func (w *Watcher) observe() {
	for {
		select {
		case ev, ok := <-w.fsEvent:
			if !ok {
				// We get here only when w.fsEvent is stopped when closing the watcher
				w.watched = nil
				w.watchers = nil
				w.dirs = nil
				close(w.fsEvent)
				w.fsEvent = nil
				return
			}
			w.parseEv(ev)
		}
	}
}
6464
func (t *peerThunk) Subscriber() peer.Subscriber {
	t.lock.RLock()
	s := t.subscriber
	t.lock.RUnlock()
	return s
}
828603
func (p *Plugin) NewWatcher(keyPrefix string) keyval.ProtoWatcher {
	return p.protoWrapper.NewWatcher(keyPrefix)
}
1330581
func (e *eventer) Subscribe() eventChannel {
	e.eventsMutex.Lock()
	defer e.eventsMutex.Unlock()
	out := make(eventChannel, eventChanBufferSize)
	e.outs[out] = out
	return out
}
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
}
902067
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),
		input:   make(chan *input, 1),
		Events:  make(chan Event, 50),
		Errors:  make(chan error),
		quit:    make(chan chan<- error, 1),
	}
	go w.readEvents()
	return w, nil
}
643907
func (c *DefaultClient) NewSubscriber(exchange, name string, durable, autoDelete bool) Subscriber {
	return &DefaultSubscriber{
		DefaultChannelClient: DefaultChannelClient{
			ctx:      c.ctx,
			client:   c,
			exchange: exchange,
			name:     "Subscriber",
		},
		name:       name,
		durable:    durable,
		autoDelete: autoDelete,
	}
}
208133
func (es *EventSystem) subscribe(sub *subscription) *Subscription {
	es.install <- sub
	<-sub.installed
	return &Subscription{ID: sub.id, f: sub, es: es}
}
376706
func (_PublicResolver *PublicResolverFilterer) WatchContentChanged(opts *bind.WatchOpts, sink chan<- *PublicResolverContentChanged, node [][32]byte) (event.Subscription, error) {

	var nodeRule []interface{}
	for _, nodeItem := range node {
		nodeRule = append(nodeRule, nodeItem)
	}

	logs, sub, err := _PublicResolver.contract.WatchLogs(opts, "ContentChanged", nodeRule)
	if err != nil {
		return nil, err
	}
	return event.NewSubscription(func(quit <-chan struct{}) error {
		defer sub.Unsubscribe()
		for {
			select {
			case log := <-logs:
				// New log arrived, parse the event and forward to the user
				event := new(PublicResolverContentChanged)
				if err := _PublicResolver.contract.UnpackLog(event, "ContentChanged", log); err != nil {
					return err
				}
				event.Raw = log

				select {
				case sink <- event:
				case err := <-sub.Err():
					return err
				case <-quit:
					return nil
				}
			case err := <-sub.Err():
				return err
			case <-quit:
				return nil
			}
		}
	}), nil
}