// SyncHandler can be wrapped around a handler to guarantee that
// only a single Log operation can proceed at a time. It's necessary
// for thread-safe concurrent writes.
966518
func SyncHandler(h Handler) Handler {
	var mu sync.Mutex
	return FuncHandler(func(r *Record) error {
		defer mu.Unlock()
		mu.Lock()
		return h.Log(r)
	})
}
1354428
func synchronizedWriters(targets ...io.Writer) []io.Writer {
	mutex := &sync.Mutex{}
	ret := make([]io.Writer, len(targets))
	for i, target := range targets {
		ret[i] = &synchronizedWriter{
			Writer: target,
			mutex:  mutex,
		}
	}
	return ret
}
64070
func (s *Server) peerHandler() {
	defer s.wg.Done()
	defer s.removeAllPeers()

	for {
		select {
		case peer := <-s.newPeers:
			s.wg.Add(1)
			go s.handleClient(peer)

		case <-s.quit:
			return
		}
	}
}
697112
func (c *TriggerController) syncResource(key string) error {
	if klog.V(4) {
		startTime := time.Now()
		klog.Infof("Started syncing resource %q", key)
		defer func() {
			klog.Infof("Finished syncing resource %q (%v)", key, time.Since(startTime))
		}()
	}

	parts := strings.SplitN(key, "/", 2)
	source := c.triggerSources[parts[0]]
	obj, exists, err := source.Store.GetByKey(parts[1])
	if err != nil {
		return fmt.Errorf("unable to retrieve %s %s from store: %v", parts[0], parts[1], err)
	}
	if !exists {
		return nil
	}

	return source.Reactor.ImageChanged(obj.(runtime.Object), c.tagRetriever)
}
469562
func (pm *ProtocolManager) syncer() {
	// Start and ensure cleanup of sync mechanisms
	pm.fetcher.Start()
	defer pm.fetcher.Stop()
	defer pm.downloader.Terminate()

	// Wait for different events to fire synchronisation operations
	forceSync := time.NewTicker(forceSyncCycle)
	defer forceSync.Stop()

	for {
		select {
		case <-pm.newPeerCh:
			// Make sure we have peers to select from, then sync
			if pm.peers.Len() < minDesiredPeerCount {
				break
			}
			go pm.synchronise(pm.peers.BestPeer())

		case <-forceSync.C:
			// Force a sync even if not enough peers are present
			go pm.synchronise(pm.peers.BestPeer())

		case <-pm.noMorePeers:
			return
		}
	}
}
418236
func (m *SyncManager) Start() {
	m.start.Do(func() {
		m.wg.Add(1)
		go m.syncerHandler()
	})
}
360848
func (c *Controller) syncHandler(key string) error {
	// Retrieve namespace and Custom Resource name from the key
	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		c.logger.Log("error", fmt.Sprintf("key '%s' is invalid: %v", key, err))
		runtime.HandleError(fmt.Errorf("key '%s' is invalid", key))
		return nil
	}

	// Custom Resource fhr contains all information we need to know about the Chart release
	fhr, err := c.fhrLister.HelmReleases(namespace).Get(name)
	if err != nil {
		if k8serrors.IsNotFound(err) {
			c.logger.Log("info", fmt.Sprintf("HelmRelease '%s' referred to in work queue no longer exists", key))
			runtime.HandleError(fmt.Errorf("HelmRelease '%s' referred to in work queue no longer exists", key))
			return nil
		}
		c.logger.Log("error", err.Error())
		return err
	}

	c.sync.ReconcileReleaseDef(*fhr)
	c.recorder.Event(fhr, corev1.EventTypeNormal, ChartSynced, MessageChartSynced)
	return nil
}
489191
func (r *Registry) AddHandler(h RegisterObject) {
	r.m.Lock()
	r.handlers[h.Reference()] = h
	r.m.Unlock()
}
1074257
func (s *Server) HTTPHandler() http.Handler {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.httpHandler
}
19347
func (p *peer) readHandler() {
	defer p.wg.Done()

	// We'll stop the timer after a new messages is received, and also
	// reset it after we process the next message.
	idleTimer := time.AfterFunc(idleTimeout, func() {
		err := fmt.Errorf("Peer %s no answer for %s -- disconnecting",
			p, idleTimeout)
		p.Disconnect(err)
	})

	// Initialize our negotiated gossip sync method before reading messages
	// off the wire. When using gossip queries, this ensures a gossip
	// syncer is active by the time query messages arrive.
	//
	// TODO(conner): have peer store gossip syncer directly and bypass
	// gossiper?
	p.initGossipSync()

	discStream := newDiscMsgStream(p)
	discStream.Start()
	defer discStream.Stop()

	chanMsgStreams := make(map[lnwire.ChannelID]*msgStream)
out:
	for atomic.LoadInt32(&p.disconnect) == 0 {
		nextMsg, err := p.readNextMessage()
		if !idleTimer.Stop() {
			select {
			case <-idleTimer.C:
			default:
			}
		}
		if err != nil {
			peerLog.Infof("unable to read message from %v: %v",
				p, err)

			switch err.(type) {
			// If this is just a message we don't yet recognize,
			// we'll continue processing as normal as this allows
			// us to introduce new messages in a forwards
			// compatible manner.
			case *lnwire.UnknownMessage:
				idleTimer.Reset(idleTimeout)
				continue

			// If they sent us an address type that we don't yet
			// know of, then this isn't a dire error, so we'll
			// simply continue parsing the remainder of their
			// messages.
			case *lnwire.ErrUnknownAddrType:
				idleTimer.Reset(idleTimeout)
				continue

			// If the NodeAnnouncement has an invalid alias, then
			// we'll log that error above and continue so we can
			// continue to read messges from the peer.
			case *lnwire.ErrInvalidNodeAlias:
				idleTimer.Reset(idleTimeout)
				continue

			// If the error we encountered wasn't just a message we
			// didn't recognize, then we'll stop all processing s
			// this is a fatal error.
			default:
				break out
			}
		}

		var (
			isChanUpdate bool
			targetChan   lnwire.ChannelID
		)

		switch msg := nextMsg.(type) {
		case *lnwire.Pong:
			// When we receive a Pong message in response to our
			// last ping message, we'll use the time in which we
			// sent the ping message to measure a rough estimate of
			// round trip time.
			pingSendTime := atomic.LoadInt64(&p.pingLastSend)
			delay := (time.Now().UnixNano() - pingSendTime) / 1000
			atomic.StoreInt64(&p.pingTime, delay)

		case *lnwire.Ping:
			pongBytes := make([]byte, msg.NumPongBytes)
			p.queueMsg(lnwire.NewPong(pongBytes), nil)

		case *lnwire.OpenChannel:
			p.server.fundingMgr.processFundingOpen(msg, p)
		case *lnwire.AcceptChannel:
			p.server.fundingMgr.processFundingAccept(msg, p)
		case *lnwire.FundingCreated:
			p.server.fundingMgr.processFundingCreated(msg, p)
		case *lnwire.FundingSigned:
			p.server.fundingMgr.processFundingSigned(msg, p)
		case *lnwire.FundingLocked:
			p.server.fundingMgr.processFundingLocked(msg, p)

		case *lnwire.Shutdown:
			select {
			case p.chanCloseMsgs <- &closeMsg{msg.ChannelID, msg}:
			case <-p.quit:
				break out
			}
		case *lnwire.ClosingSigned:
			select {
			case p.chanCloseMsgs <- &closeMsg{msg.ChannelID, msg}:
			case <-p.quit:
				break out
			}

		case *lnwire.Error:
			key := p.addr.IdentityKey

			switch {
			// In the case of an all-zero channel ID we want to
			// forward the error to all channels with this peer.
			case msg.ChanID == lnwire.ConnectionWideID:
				for chanID, chanStream := range chanMsgStreams {
					chanStream.AddMsg(nextMsg)

					// Also marked this channel as failed,
					// so we won't try to restart it on
					// reconnect with this peer.
					p.failedChannels[chanID] = struct{}{}
				}

			// If the channel ID for the error message corresponds
			// to a pending channel, then the funding manager will
			// handle the error.
			case p.server.fundingMgr.IsPendingChannel(msg.ChanID, key):
				p.server.fundingMgr.processFundingError(msg, key)

			// If not we hand the error to the channel link for
			// this channel.
			default:
				isChanUpdate = true
				targetChan = msg.ChanID

				// Also marked this channel as failed, so we
				// won't try to restart it on reconnect with
				// this peer.
				p.failedChannels[targetChan] = struct{}{}
			}

		// TODO(roasbeef): create ChanUpdater interface for the below
		case *lnwire.UpdateAddHTLC:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.UpdateFulfillHTLC:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.UpdateFailMalformedHTLC:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.UpdateFailHTLC:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.RevokeAndAck:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.CommitSig:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.UpdateFee:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.ChannelReestablish:
			isChanUpdate = true
			targetChan = msg.ChanID

		case *lnwire.ChannelUpdate,
			*lnwire.ChannelAnnouncement,
			*lnwire.NodeAnnouncement,
			*lnwire.AnnounceSignatures,
			*lnwire.GossipTimestampRange,
			*lnwire.QueryShortChanIDs,
			*lnwire.QueryChannelRange,
			*lnwire.ReplyChannelRange,
			*lnwire.ReplyShortChanIDsEnd:

			discStream.AddMsg(msg)

		default:
			peerLog.Errorf("unknown message %v received from peer "+
				"%v", uint16(msg.MsgType()), p)
		}

		if isChanUpdate {
			// If this is a channel update, then we need to feed it
			// into the channel's in-order message stream.
			chanStream, ok := chanMsgStreams[targetChan]
			if !ok {
				// If a stream hasn't yet been created, then
				// we'll do so, add it to the map, and finally
				// start it.
				chanStream = newChanMsgStream(p, targetChan)
				chanMsgStreams[targetChan] = chanStream
				chanStream.Start()
				defer chanStream.Stop()
			}

			// With the stream obtained, add the message to the
			// stream so we can continue processing message.
			chanStream.AddMsg(nextMsg)
		}

		idleTimer.Reset(idleTimeout)
	}

	p.Disconnect(errors.New("read handler closed"))

	peerLog.Tracef("readHandler for peer %v done", p)
}
402937
func (p *chainMonitor) BlockConnectedHandler() {
	defer p.wg.Done()
out:
	for {
	keepon:
		select {
		case hash, ok := <-p.blockChan:
			if !ok {
				log.Warnf("Block connected channel closed.")
				break out
			}

			// Do not handle reorg and block connects simultaneously.
			p.reorgLock.Lock()
			// Collect block data.
			msgBlock, blockData, err := p.collect(hash)
			p.reorgLock.Unlock()
			if err != nil {
				log.Errorf("Failed to collect data for block %v: %v", hash, err)
				break keepon
			}

			// Store block data with each saver.
			for _, s := range p.dataSavers {
				if s != nil {
					// Save data to wherever the saver wants to put it.
					if err = s.Store(blockData, msgBlock); err != nil {
						log.Errorf("(%v).Store failed: %v", reflect.TypeOf(s), err)
					}
				}
			}

		case <-p.ctx.Done():
			log.Debugf("Got quit signal. Exiting block connected handler.")
			break out
		}
	}

}
1101775
func (c *Client) workHandler() {
	for {
		select {
		case w := <-c.work:
			c.process(w)
		case <-time.After(c.keepAlive):
			c.process(keepAliveData)
		case <-c.disconnect:
			return
		}
	}
}
1285702
func (n *Negroni) UseHandler(handler http.Handler) {
	n.Use(Wrap(handler))
}
507404
func (ctx *context) NextHandler() Handler {
	if ctx.IsStopped() {
		return nil
	}
	nextIndex := ctx.currentHandlerIndex + 1
	// check if it has a next middleware
	if nextIndex < len(ctx.handlers) {
		return ctx.handlers[nextIndex]
	}
	return nil
}
1029119
func MultiHandler(handlers ...Handler) Handler {
	c := make([]Handler, len(handlers))
	copy(c, handlers)
	return &multiHandler{
		handlers: c,
	}
}
1247126
func (n *Node) RPCHandler() (*rpc.Server, error) {
	n.lock.RLock()
	defer n.lock.RUnlock()

	if n.inprocHandler == nil {
		return nil, ErrNodeStopped
	}
	return n.inprocHandler, nil
}
573941
func (app *App) UseHandler(h Handler) *App {
	app.mds = append(app.mds, h.Serve)
	return app
}