// watchPeerEvents reads peer events from the given channel and emits
// corresponding network events
509292
func (net *Network) watchPeerEvents(id enode.ID, events chan *p2p.PeerEvent, sub event.Subscription) {
	defer func() {
		sub.Unsubscribe()

		// assume the node is now down
		net.lock.Lock()
		defer net.lock.Unlock()

		node := net.getNode(id)
		if node == nil {
			return
		}
		node.SetUp(false)
		ev := NewEvent(node)
		net.events.Send(ev)
	}()
	for {
		select {
		case event, ok := <-events:
			if !ok {
				return
			}
			peer := event.Peer
			switch event.Type {

			case p2p.PeerEventTypeAdd:
				net.DidConnect(id, peer)

			case p2p.PeerEventTypeDrop:
				net.DidDisconnect(id, peer)

			case p2p.PeerEventTypeMsgSend:
				net.DidSend(id, peer, event.Protocol, *event.MsgCode)

			case p2p.PeerEventTypeMsgRecv:
				net.DidReceive(peer, id, event.Protocol, *event.MsgCode)

			}

		case err := <-sub.Err():
			if err != nil {
				log.Error("Error in peer event subscription", "id", id, "err", err)
			}
			return
		}
	}
}
117046
func (tn *TreeNode) processWatchEvent(evt *zk.Event) {
	tn.tree.logger.Debugf("ProcessWatchEvent: %v", evt)
	switch evt.Type {
	case zk.EventNodeCreated:
		if tn.parent != nil {
			tn.tree.handleException(errors.New("unexpected NodeCreated on non-root node"))
			return
		}
		tn.wasCreated()
	case zk.EventNodeChildrenChanged:
		tn.refreshChildren()
	case zk.EventNodeDataChanged:
		tn.refreshData()
	case zk.EventNodeDeleted:
		tn.wasDeleted()
	default:
		// Leave other type of events unhandled
		// tn.tree.logger.Printf("Event received: %v", evt)
	}
}
795466
func (pl *List) waitForPeerAddedEvent(ctx context.Context) error {
	if _, ok := ctx.Deadline(); !ok {
		return pl.newNoContextDeadlineError()
	}

	select {
	case <-pl.peerAvailableEvent:
		return nil
	case <-ctx.Done():
		return pl.newUnavailableError(ctx.Err())
	}
}
1200735
func (mock *Mock) Watch(paths []string, onEvent func(event fsnotify.Event), onClose func()) error {
	go func() {
		for {
			select {
			case event, ok := <-mock.eventChan:
				if !ok {
					onClose()
					return
				}
				onEvent(event)
			}
		}
	}()
	return 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)
	}
}
264985
func (self *User) Watch(eventsChannel *EventChannel) error {
	self.stop = make(chan struct{})
	go func() {
		for {
			select {
			case event := <-eventsChannel.channel:
				err := self.alert(event)
				if err != nil {
					log.Warningf("Failed to process watch event: %v", err)
				}
			case <-self.stop:
				log.Info("user watcher exiting")
				return
			}
		}
	}()
	return nil
}
1158346
func (ew *EventsWriter) WatchForEvents(request *Request) (*EventChannel, error) {
	if ew.H != nil {
		return ew.H.WatchEvents(request)
	}
	return nil, nil
}
1276630
func WatchRepo(userID, repoID int64, watch bool) (err error) {
	return watchRepo(x, userID, repoID, watch)
}
300615
func (conR *ConsensusReactor) subscribeToBroadcastEvents() {
	const subscriber = "consensus-reactor"
	conR.conS.evsw.AddListenerForEvent(subscriber, types.EventNewRoundStep,
		func(data tmevents.EventData) {
			conR.broadcastNewRoundStepMessage(data.(*cstypes.RoundState))
		})

	conR.conS.evsw.AddListenerForEvent(subscriber, types.EventValidBlock,
		func(data tmevents.EventData) {
			conR.broadcastNewValidBlockMessage(data.(*cstypes.RoundState))
		})

	conR.conS.evsw.AddListenerForEvent(subscriber, types.EventVote,
		func(data tmevents.EventData) {
			conR.broadcastHasVoteMessage(data.(*types.Vote))
		})

}
1230994
func WatchEgressAddressesForRelations(resources facade.Resources, st State, relations params.Entities) (params.StringsWatchResults, error) {
	results := params.StringsWatchResults{
		make([]params.StringsWatchResult, len(relations.Entities)),
	}

	one := func(tag string) (id string, changes []string, _ error) {
		logger.Debugf("Watching egress addresses for %+v", tag)

		relationTag, err := names.ParseRelationTag(tag)
		if err != nil {
			return "", nil, errors.Trace(err)
		}

		// Load the relation details for the current token.
		localEndpoint, err := localApplication(st, relationTag)
		if err != nil {
			return "", nil, errors.Trace(err)
		}

		w, err := NewEgressAddressWatcher(st, localEndpoint.relation, localEndpoint.application)
		if err != nil {
			return "", nil, errors.Trace(err)
		}

		// TODO(wallyworld) - we will need to watch subnets too, but only
		// when we support using cloud local addresses
		//filter := func(id interface{}) bool {
		//	include, err := includeAsIngressSubnet(id.(string))
		//	if err != nil {
		//		logger.Warningf("invalid CIDR %q", id)
		//	}
		//	return include
		//}
		//w := api.st.WatchSubnets(filter)

		changes, ok := <-w.Changes()
		if !ok {
			return "", nil, common.ServerError(watcher.EnsureErr(w))
		}
		return resources.Register(w), changes, nil
	}

	for i, e := range relations.Entities {
		watcherId, changes, err := one(e.Tag)
		if err != nil {
			results.Results[i].Error = common.ServerError(err)
			continue
		}
		results.Results[i].StringsWatcherId = watcherId
		results.Results[i].Changes = changes
	}
	return results, nil
}
788793
func (r *Relation) WatchRelationIngressNetworks() StringsWatcher {
	return newrelationNetworksWatcher(r.st, r.Tag().Id(), ingress)
}
948542
func (m *Machine) WatchForRebootEvent() NotifyWatcher {
	machineIds := m.machinesToCareAboutRebootsFor()
	machines := set.NewStrings(machineIds...)

	filter := func(key interface{}) bool {
		if id, ok := key.(string); ok {
			if id, err := m.st.strictLocalID(id); err == nil {
				return machines.Contains(id)
			} else {
				return false
			}
		}
		return false
	}
	return newNotifyCollWatcher(m.st, rebootC, filter)
}
212365
func (feed *TapFeed) forwardTapEvents(singleFeed *memcached.TapFeed, killSwitch chan bool, host string) {
	defer feed.wg.Done()
	for {
		select {
		case event, ok := <-singleFeed.C:
			if !ok {
				if singleFeed.Error != nil {
					logging.Errorf("go-couchbase: Tap feed from %s failed: %v", host, singleFeed.Error)
				}
				killSwitch <- true
				return
			}
			feed.output <- event
		case <-feed.quit:
			return
		}
	}
}
975789
func PeepEvents(events []Event, action EventAction, minType, maxType uint32) (storedEvents int, err error) {
	if events == nil {
		return 0, nil
	}

	var _events []CEvent = make([]CEvent, len(events))

	if action == ADDEVENT { // the contents of _events matter if they are to be added
		for i := 0; i < len(events); i++ {
			_events[i] = *cEvent(events[i])
		}
	}

	_pevents := (*C.SDL_Event)(unsafe.Pointer(&_events[0]))
	storedEvents = int(C.SDL_PeepEvents(_pevents, C.int(len(events)), action.c(), C.Uint32(minType), C.Uint32(maxType)))

	if action != ADDEVENT { // put events into slice, events unchanged if action = ADDEVENT
		for i := 0; i < storedEvents; i++ {
			events[i] = goEvent(&_events[i])
		}
	}

	if storedEvents < 0 {
		err = GetError()
	}

	return
}
1151989
func (d *EtcdStateDriver) WatchAll(baseKey string, rsps chan [2][]byte) error {
	watcher := d.KeysAPI.Watcher(baseKey, &client.WatcherOptions{Recursive: true})
	if watcher == nil {
		log.Errorf("etcd watch failed.")
		return errors.New("etcd watch failed")
	}

	go d.channelEtcdEvents(watcher, rsps)

	return nil
}
816939
func eventWatch(params map[string]interface{}) (WatcherFunc, error) {
	// The stale setting doesn't apply to events.

	var name string
	if err := assignValue(params, "name", &name); err != nil {
		return nil, err
	}

	fn := func(p *Plan) (BlockingParamVal, interface{}, error) {
		event := p.client.Event()
		opts := makeQueryOptionsWithContext(p, false)
		defer p.cancelFunc()
		events, meta, err := event.List(name, &opts)
		if err != nil {
			return nil, nil, err
		}

		// Prune to only the new events
		for i := 0; i < len(events); i++ {
			if WaitIndexVal(event.IDToIndex(events[i].ID)).Equal(p.lastParamVal) {
				events = events[i+1:]
				break
			}
		}
		return WaitIndexVal(meta.LastIndex), events, err
	}
	return fn, nil
}
481239
func (r *registry) WatchSubnets(since uint64, stop chan bool) (*subnet.Response, error) {
	for {
		select {
		case event, ok := <-r.events:
			if !ok {
				return nil, errors.New("unexpected close of discoverd event stream")
			}
			if event.Kind != discoverd.EventKindServiceMeta {
				continue
			}
			net := &Network{}
			if err := json.Unmarshal(event.ServiceMeta.Data, net); err != nil {
				return nil, err
			}
			subnets := make(map[string][]byte)
			for subnet, data := range net.Subnets {
				if known, ok := knownSubnets[subnet]; ok && reflect.DeepEqual(known, data) {
					continue
				}
				subnets[subnet] = []byte(*data)
			}
			knownSubnets = net.Subnets
			if event.ServiceMeta.Index >= since {
				return &subnet.Response{Subnets: subnets, Index: event.ServiceMeta.Index}, nil
			}
		case <-stop:
			return nil, nil
		}
	}
}