// notifyBatchLocked sends out a notification for all the ops in md.
559883
func (fbo *folderBranchOps) notifyBatchLocked(
	ctx context.Context, lState *kbfssync.LockState,
	md ImmutableRootMetadata) error {
	fbo.headLock.AssertLocked(lState)

	for _, op := range md.data.Changes.Ops {
		err := fbo.notifyOneOpLocked(ctx, lState, op, md.ReadOnly(), false)
		if err != nil {
			return err
		}
	}
	return nil
}
303931
func (k KeybaseServiceMeasured) NotifyOverallSyncStatus(
	ctx context.Context, status keybase1.FolderSyncStatus) (err error) {
	k.notifyTimer.Time(func() {
		err = k.delegate.NotifyOverallSyncStatus(ctx, status)
	})
	return err
}
733752
func (t *peerThunk) NotifyStatusChanged(pid peer.Identifier) {
	t.list.notifyStatusChanged(pid)

	s := t.Subscriber()
	if s != nil {
		s.NotifyStatusChanged(pid)
	}
}
87261
func (l *State) StopNotifyProxy(ch chan<- struct{}) {
	l.Lock()
	defer l.Unlock()
	delete(l.managedProxyHandlers, ch)
}
901005
func (k *KBPKIClient) Notify(ctx context.Context, notification *keybase1.FSNotification) error {
	return k.serviceOwner.KeybaseService().Notify(ctx, notification)
}
701968
func (tx *transaction) notifyActiveIters() {
	tx.activeIterLock.RLock()
	for _, iter := range tx.activeIters {
		iter.ForceReseek()
	}
	tx.activeIterLock.RUnlock()
}
1096423
func (m *scopedManager) CloseNotify() <-chan bool {
	if w, ok := m.ResponseWriter.(http.CloseNotifier); ok {
		return w.CloseNotify()
	}
	return nil
}
953754
func notifyBlock(chain ChainSource, ro *rescanOptions,
	curHeader wire.BlockHeader, curStamp waddrmgr.BlockStamp,
	scanning bool) error {

	// Find relevant transactions based on watch list. If scanning is
	// false, we can safely assume this block has no relevant transactions.
	var relevantTxs []*btcutil.Tx
	if len(ro.watchList) != 0 && scanning {
		// If we have a non-empty watch list, then we need to see if it
		// matches the rescan's filters, so we get the basic filter
		// from the DB or network.
		matched, err := blockFilterMatches(chain, ro, &curStamp.Hash)
		if err != nil {
			return err
		}

		if matched {
			relevantTxs, err = extractBlockMatches(
				chain, ro, &curStamp,
			)
			if err != nil {
				return err
			}
		}
	}

	if ro.ntfn.OnFilteredBlockConnected != nil {
		ro.ntfn.OnFilteredBlockConnected(curStamp.Height, &curHeader,
			relevantTxs)
	}

	if ro.ntfn.OnBlockConnected != nil {
		ro.ntfn.OnBlockConnected(&curStamp.Hash,
			curStamp.Height, curHeader.Timestamp)
	}

	return nil
}
126332
func (ai *Amqp) NotifyClose(c chan *amqp.Error) chan *amqp.Error {
	return ai.conn.NotifyClose(c)
}
111681
func (influxdb *InfluxdbNotifier) Notify(messages Messages) bool {
	config := &client.ClientConfig{
		Host:     influxdb.Host,
		Username: influxdb.Username,
		Password: influxdb.Password,
		Database: influxdb.Database,
	}

	influxdbClient, err := client.New(config)
	if err != nil {
		log.Println("unable to access influxdb. can't send notification. ", err)
		return false
	}

	seriesList := influxdb.toSeries(messages)
	err = influxdbClient.WriteSeries(seriesList)

	if err != nil {
		log.Println("unable to send notifications: ", err)
		return false
	}

	log.Println("influxdb notification sent.")
	return true
}
503874
func RetryNotify(operation Operation, b Backoff, notify Notify) error {
	var err error
	var wait time.Duration
	var retry bool
	var n int

	for {
		if err = operation(); err == nil {
			return nil
		}

		n++
		wait, retry = b.Next(n)
		if !retry {
			return err
		}

		if notify != nil {
			notify(err)
		}

		time.Sleep(wait)
	}
}
1019873
func (a *aggregator) notify(oldEvents map[string]*aggregatedEvent, out chan<- []string) {
	timeBeforeSending := time.Now()
	l.Debugf("%v Notifying about %d fs events", a, len(oldEvents))
	separatedBatches := make(map[fs.EventType][]string)
	for path, event := range oldEvents {
		separatedBatches[event.evType] = append(separatedBatches[event.evType], path)
	}
	for _, evType := range [3]fs.EventType{fs.NonRemove, fs.Mixed, fs.Remove} {
		currBatch := separatedBatches[evType]
		if len(currBatch) != 0 {
			select {
			case out <- currBatch:
			case <-a.ctx.Done():
				return
			}
		}
	}
	// If sending to channel blocked for a long time,
	// shorten next notifyDelay accordingly.
	duration := time.Since(timeBeforeSending)
	buffer := time.Millisecond
	var nextDelay time.Duration
	switch {
	case duration < a.notifyDelay/10:
		nextDelay = a.notifyDelay
	case duration+buffer > a.notifyDelay:
		nextDelay = buffer
	default:
		nextDelay = a.notifyDelay - duration
	}
	select {
	case a.notifyTimerResetChan <- nextDelay:
	case <-a.ctx.Done():
	}
}
142922
func (c *Conn) notify(method string, data []byte) {
	c.streamMu.Lock()
	stream := c.streams[method]
	c.streamMu.Unlock()

	if stream != nil {
		// Stream writer must be able to handle incoming writes
		// even after it has been removed (unsubscribed).
		stream.write(method, data)
	}
}
591899
func (m *SubscriptionManager) notifySubscriber(sub *newSubscription,
	block BlockNtfn) {

	select {
	case sub.ntfnQueue.ChanIn() <- block:
	case <-sub.quit:
	case <-m.quit:
		return
	}
}
1356968
func (s *NeutrinoClient) notificationHandler() {
	hash, height, err := s.GetBestBlock()
	if err != nil {
		log.Errorf("Failed to get best block from chain service: %s",
			err)
		s.Stop()
		s.wg.Done()
		return
	}

	bs := &waddrmgr.BlockStamp{Hash: *hash, Height: height}

	// TODO: Rather than leaving this as an unbounded queue for all types of
	// notifications, try dropping ones where a later enqueued notification
	// can fully invalidate one waiting to be processed.  For example,
	// blockconnected notifications for greater block heights can remove the
	// need to process earlier blockconnected notifications still waiting
	// here.

	var notifications []interface{}
	enqueue := s.enqueueNotification
	var dequeue chan interface{}
	var next interface{}
out:
	for {
		s.clientMtx.Lock()
		rescanErr := s.rescanErr
		s.clientMtx.Unlock()
		select {
		case n, ok := <-enqueue:
			if !ok {
				// If no notifications are queued for handling,
				// the queue is finished.
				if len(notifications) == 0 {
					break out
				}
				// nil channel so no more reads can occur.
				enqueue = nil
				continue
			}
			if len(notifications) == 0 {
				next = n
				dequeue = s.dequeueNotification
			}
			notifications = append(notifications, n)

		case dequeue <- next:
			if n, ok := next.(BlockConnected); ok {
				bs = &waddrmgr.BlockStamp{
					Height: n.Height,
					Hash:   n.Hash,
				}
			}

			notifications[0] = nil
			notifications = notifications[1:]
			if len(notifications) != 0 {
				next = notifications[0]
			} else {
				// If no more notifications can be enqueued, the
				// queue is finished.
				if enqueue == nil {
					break out
				}
				dequeue = nil
			}

		case err := <-rescanErr:
			if err != nil {
				log.Errorf("Neutrino rescan ended with error: %s", err)
			}

		case s.currentBlock <- bs:

		case <-s.quit:
			break out
		}
	}

	s.Stop()
	close(s.dequeueNotification)
	s.wg.Done()
}
362854
func (s *routeStats) Notify(c context.Context, req *RouteTrace) error {
	key := routeKey{
		Method:     req.Method,
		Route:      req.Route,
		StatusCode: req.StatusCode,
		Time:       req.StartTime.UTC().Truncate(time.Minute),
	}

	s.mu.Lock()
	s.init()
	stat, ok := s.m[key]
	if !ok {
		stat = &routeStat{}
		s.m[key] = stat
	}
	addWG := s.addWG
	addWG.Add(1)
	s.mu.Unlock()

	dur := req.EndTime.Sub(req.StartTime)

	stat.mu.Lock()
	err := stat.Add(dur)
	addWG.Done()
	stat.mu.Unlock()

	return err
}
741841
func (c *BitcoindClient) NotifyReceived(addrs []btcutil.Address) error {
	c.NotifyBlocks()

	select {
	case c.rescanUpdate <- addrs:
	case <-c.quit:
		return ErrBitcoindClientShuttingDown
	}

	return nil
}