// readHandler is responsible for reading messages off the wire in series, then
// properly dispatching the handling of the message to the proper subsystem.
//
// NOTE: This method MUST be run as a goroutine.
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)
}
926592
func (r *FlatChain) ChainHandler(h http.Handler) FlatChains {
	fh := FlatChainHandlerWrap(h, r.log)
	r.Chain(fh)
	return fh
}
451994
func (s *Swarm) StreamHandler() inet.StreamHandler {
	handler, _ := s.streamh.Load().(inet.StreamHandler)
	return handler
}
905594
func (m *Messenger) HandleMessage(f MessageHandler) {
	m.messageHandlers = append(m.messageHandlers, f)
}
1293718
func (c *Conn) readErr(timeout bool, e error) *net.OpError {
	return &net.OpError{
		Op:   "read",
		Net:  network,
		Addr: c.RemoteAddr(),
		Err:  neterr{timeout: timeout, err: e},
	}
}
282166
func (c *central) handleReadByType(b []byte) []byte {
	start, end := readHandleRange(b[:4])
	t := UUID{b[4:]}

	w := newL2capWriter(c.mtu)
	w.WriteByteFit(attOpReadByTypeRsp)
	uuidLen := -1
	for _, a := range c.attrs.Subrange(start, end) {
		if !a.typ.Equal(t) {
			continue
		}
		if (a.secure&CharRead) != 0 && c.security > securityLow {
			return attErrorRsp(attOpReadByTypeReq, start, attEcodeAuthentication)
		}
		v := a.value
		if v == nil {
			rsp := newResponseWriter(int(c.mtu - 1))
			req := &ReadRequest{
				Request: Request{Central: c},
				Cap:     int(c.mtu - 1),
				Offset:  0,
			}
			if c, ok := a.pvt.(*Characteristic); ok {
				c.rhandler.ServeRead(rsp, req)
			} else if d, ok := a.pvt.(*Descriptor); ok {
				d.rhandler.ServeRead(rsp, req)
			}
			v = rsp.bytes()
		}
		if uuidLen == -1 {
			uuidLen = len(v)
			w.WriteByteFit(byte(uuidLen) + 2)
		}
		if len(v) != uuidLen {
			break
		}
		w.Chunk()
		w.WriteUint16Fit(a.h)
		w.WriteFit(v)
		if ok := w.Commit(); !ok {
			break
		}
	}
	if uuidLen == -1 {
		return attErrorRsp(attOpReadByTypeReq, start, attEcodeAttrNotFound)
	}
	return w.Bytes()
}
31977
func handleStdLogger(done chan<- struct{}) {
	r, w := io.Pipe()
	defer func() {
		_ = r.Close()
		_ = w.Close()
	}()

	stdlog.SetOutput(w)

	scanner := bufio.NewScanner(r)

	go func() {
		done <- struct{}{}
	}()

	for scanner.Scan() {
		log.WithField("stdlog", true).Info(scanner.Text())
	}
}
606392
func handleErr(r jsonResponse) error {
	if !r.Success {
		return errors.New(r.Message)
	}
	return nil
}
1249796
func (s *Server) Handle(c *Connection) {
	s.mutex.Lock()
	defer s.mutex.Unlock()

	s.messages = append(s.messages, "")

	s.readHello(c)
	s.readSender(c)
	s.readRecipients(c)
	s.readData(c)

	c.conn.Close()
}
1240310
func (r *ReadContext) ReadMessage(msg proto.Message) error {
	savior.Debugf("wire.ReadContext: Reading message at %d", r.offset)

	length, err := binary.ReadUvarint(r.countingReader)
	if err != nil {
		return errors.WithStack(err)
	}

	msgBuf := r.protoBuffer.Bytes()
	if cap(msgBuf) < int(length) {
		msgBuf = make([]byte, nextPowerOf2(int(length)))
	}

	_, err = io.ReadFull(r.countingReader, msgBuf[:length])
	if err != nil {
		return errors.WithStack(err)
	}
	r.protoBuffer.SetBuf(msgBuf[:length])

	msg.Reset()

	err = r.protoBuffer.Unmarshal(msg)
	if err != nil {
		return errors.WithStack(err)
	}

	if DebugWire {
		fmt.Printf(">> %s %+v\n", reflect.TypeOf(msg).Elem().Name(), msg)
	}

	return nil
}
1348555
func DeregisterReaderHandler(name string) {
	readerRegisterLock.Lock()
	delete(readerRegister, name)
	readerRegisterLock.Unlock()
}
1064646
func (sp *ServerPeer) OnRead(_ *peer.Peer, bytesRead int, msg wire.Message,
	err error) {

	sp.server.AddBytesReceived(uint64(bytesRead))

	// Send a message to each subscriber. Each message gets its own
	// goroutine to prevent blocking on the mutex lock.
	// TODO: Flood control.
	sp.mtxSubscribers.RLock()
	defer sp.mtxSubscribers.RUnlock()
	for subscription := range sp.recvSubscribers {
		go func(subscription spMsgSubscription) {
			select {
			case <-subscription.quitChan:
			case subscription.msgChan <- spMsg{
				msg: msg,
				sp:  sp,
			}:
			}
		}(subscription)
	}
}
225755
func (ref *Consumer) messageHandler(in <-chan *sarama.ConsumerMessage) {
	ref.Debug("messageHandler started ...")
	var prevValue []byte

	for {
		select {
		case msg := <-in:
			if msg == nil {
				continue
			}
			consumerMsg := &ConsumerMessage{
				Key:       msg.Key,
				Value:     msg.Value,
				PrevValue: prevValue,
				Topic:     msg.Topic,
				Partition: msg.Partition,
				Offset:    msg.Offset,
				Timestamp: msg.Timestamp,
			}
			// Store value as previous for the next iteration
			prevValue = consumerMsg.Value
			select {
			case ref.Config.RecvMessageChan <- consumerMsg:
			case <-time.After(1 * time.Second):
				ref.Warn("Failed to deliver a message")
			}
		case <-ref.closeChannel:
			ref.Debug("Canceling message handler")
			return
		}
	}
}
1181142
func (r *Router) UseHandler(handler http.Handler) *Router {
	r.Route("/*").All(handler.ServeHTTP)
	return r
}
222198
func GetHandler(ctx *router.Context) {
	start := clock.Now(ctx.Context)
	// Start the fetcher and wait for fetched logs to arrive into ch.
	data, err := startFetch(ctx.Context, ctx.Request, ctx.Params.ByName("path"))
	if err != nil {
		logging.WithError(err).Errorf(ctx.Context, "failed to start fetch")
		writeErrorPage(ctx, err, data)
		return
	}
	writeOKHeaders(ctx, data)

	// Write the log contents and then the footer.
	err = serve(ctx.Context, data, ctx.Writer)
	if err != nil {
		logging.WithError(err).Errorf(ctx.Context, "failed to serve logs")
	}
	writeFooter(ctx, start, err, data.options.isHTML())
}
275117
func (s *StreamingRpcRegistry) GetHandler(method string) (StreamingRpcHandler, error) {
	h, ok := s.registry[method]
	if !ok {
		return nil, fmt.Errorf("%s: %q", ErrUnknownMethod, method)
	}

	return h, nil
}
743694
func (r *Reader) ReadMessage() (*Message, error) {
	line, err := r.reader.ReadString('\n')
	if err != nil {
		return nil, err
	}

	if r.DebugCallback != nil {
		r.DebugCallback(line)
	}

	// Parse the message from our line
	return ParseMessage(line)
}