Method processMsg

nats.go:3670–3904

processMsg is called by parse and will place the msg on the appropriate channel/pending queue for processing. If the channel is full, or the pending queue is over the pending limits, the connection is considered a slow consumer.

(data []byte)

// or the pending queue is over the pending limits, the connection is
// considered a slow consumer.
func (nc *Conn) processMsg(data []byte) {
	// Stats
	atomic.AddUint64(&nc.InMsgs, 1)
	atomic.AddUint64(&nc.InBytes, uint64(len(data)))

	// Don't lock the connection to avoid server cutting us off if the
	// flusher is holding the connection lock, trying to send to the server
	// that is itself trying to send data to us.
	nc.subsMu.RLock()
	sub := nc.subs[nc.ps.ma.sid]
	var mf msgFilter
	if nc.filters != nil {
		mf = nc.filters[string(nc.ps.ma.subject)]
	}
	nc.subsMu.RUnlock()

	if sub == nil {
		return
	}

	// Copy them into string
	subj := string(nc.ps.ma.subject)
	reply := string(nc.ps.ma.reply)

	// Doing message create outside of the sub's lock to reduce contention.
	// It's possible that we end-up not using the message, but that's ok.

	// FIXME(dlc): Need to copy, should/can do COW?
	msgPayload := data
	if !nc.ps.msgCopied {
		msgPayload = make([]byte, len(data))
		copy(msgPayload, data)
	}

	// Check if we have headers encoded here.
	var h Header
	var err error
	var ctrlMsg bool
	var ctrlType int
	var fcReply string

	if nc.ps.ma.hdr > 0 {
		hbuf := msgPayload[:nc.ps.ma.hdr]
		msgPayload = msgPayload[nc.ps.ma.hdr:]
		h, err = DecodeHeadersMsg(hbuf)
		if err != nil {
			// We will pass the message through but send async error.
			nc.mu.Lock()
			nc.err = ErrBadHeaderMsg
			if asyncErrorCB := nc.Opts.AsyncErrorCB; asyncErrorCB != nil {
				nc.ach.push(func() { asyncErrorCB(nc, sub, ErrBadHeaderMsg) })
			}
			nc.mu.Unlock()
		}
	}

	// FIXME(dlc): Should we recycle these containers?
	m := &Msg{