// deliverMsg will deliver a message to a matching subscription and its underlying client.
// We process all connection/client types. mh is the part that will be protocol/client specific.
321897
func (c *client) deliverMsg(sub *subscription, mh, msg []byte) bool {
	if sub.client == nil {
		return false
	}
	client := sub.client
	client.mu.Lock()

	// Check echo
	if c == client && !client.echo {
		client.mu.Unlock()
		return false
	}

	// Check if we have a subscribe deny clause. This will trigger us to check the subject
	// for a match against the denied subjects.
	if client.mperms != nil && client.checkDenySub(string(c.pa.subject)) {
		client.mu.Unlock()
		return false
	}

	srv := client.srv

	sub.nm++
	// Check if we should auto-unsubscribe.
	if sub.max > 0 {
		if client.kind == ROUTER && sub.nm >= sub.max {
			// The only router based messages that we will see here are remoteReplies.
			// We handle these slightly differently.
			defer client.removeReplySub(sub)
		} else {
			// For routing..
			shouldForward := client.kind == CLIENT || client.kind == SYSTEM && client.srv != nil
			// If we are at the exact number, unsubscribe but
			// still process the message in hand, otherwise
			// unsubscribe and drop message on the floor.
			if sub.nm == sub.max {
				client.Debugf("Auto-unsubscribe limit of %d reached for sid '%s'", sub.max, string(sub.sid))
				// Due to defer, reverse the code order so that execution
				// is consistent with other cases where we unsubscribe.
				if shouldForward {
					defer srv.updateRouteSubscriptionMap(client.acc, sub, -1)
				}
				defer client.unsubscribe(client.acc, sub, true)
			} else if sub.nm > sub.max {
				client.Debugf("Auto-unsubscribe limit [%d] exceeded", sub.max)
				client.mu.Unlock()
				client.unsubscribe(client.acc, sub, true)
				if shouldForward {
					srv.updateRouteSubscriptionMap(client.acc, sub, -1)
				}
				return false
			}
		}
	}

	// Update statistics

	// The msg includes the CR_LF, so pull back out for accounting.
	msgSize := int64(len(msg) - LEN_CR_LF)

	// No atomic needed since accessed under client lock.
	// Monitor is reading those also under client's lock.
	client.outMsgs++
	client.outBytes += msgSize

	atomic.AddInt64(&srv.outMsgs, 1)
	atomic.AddInt64(&srv.outBytes, msgSize)

	// Check for internal subscription.
	if client.kind == SYSTEM {
		s := client.srv
		client.mu.Unlock()
		s.deliverInternalMsg(sub, c.pa.subject, c.pa.reply, msg[:msgSize])
		return true
	}

	// If we are a client and we detect that the consumer we are
	// sending to is in a stalled state, go ahead and wait here
	// with a limit.
	if c.kind == CLIENT && client.out.stc != nil {
		client.stalledWait(c)
	}

	// Check for closed connection
	if client.flags.isSet(clearConnection) {
		client.mu.Unlock()
		return false
	}

	// Queue to outbound buffer
	client.queueOutbound(mh)
	client.queueOutbound(msg)

	client.out.pm++

	// Check outbound threshold and queue IO flush if needed.
	// This is specifically looking at situations where we are getting behind and may want
	// to intervene before this producer goes back to top of readloop. We are in the producer's
	// readloop go routine at this point.
	// FIXME(dlc) - We may call this alot, maybe suppress after first call?
	if client.out.pm > 1 && client.out.pb > maxBufSize*2 {
		client.flushSignal()
	}

	// Add the data size we are responsible for here. This will be processed when we
	// return to the top of the readLoop.
	if _, ok := c.pcd[client]; !ok {
		client.out.fsp++
		c.pcd[client] = needFlush
	}

	if c.trace {
		client.traceOutOp(string(mh[:len(mh)-LEN_CR_LF]), nil)
	}

	client.mu.Unlock()

	return true
}
105793
func NewMsg(code byte, name string, payload []byte) *Msg {
	return &Msg{
		Code:       code,
		Name:       []byte(name),
		Payload:    payload,
		namestring: name,
	}
}
543803
func (broker *Broker) WorkerMsg(sender string, msg []string) {
	//  At least, command
	if len(msg) == 0 {
		panic("len(msg) == 0")
	}

	command, msg := popStr(msg)
	id_string := fmt.Sprintf("%q", sender)
	_, worker_ready := broker.workers[id_string]
	worker := broker.WorkerRequire(sender)

	switch command {
	case mdapi.MDPW_READY:
		if worker_ready { //  Not first command in session
			worker.Delete(true)
		} else if len(sender) >= 4 /*  Reserved service name */ && sender[:4] == "mmi." {
			worker.Delete(true)
		} else {
			//  Attach worker to service and mark as idle
			worker.service = broker.ServiceRequire(msg[0])
			worker.Waiting()
		}
	case mdapi.MDPW_REPLY:
		if worker_ready {
			//  Remove & save client return envelope and insert the
			//  protocol header and service name, then rewrap envelope.
			client, msg := unwrap(msg)
			broker.socket.SendMessage(client, "", mdapi.MDPC_CLIENT, worker.service.name, msg)
			worker.Waiting()
		} else {
			worker.Delete(true)
		}
	case mdapi.MDPW_HEARTBEAT:
		if worker_ready {
			worker.expiry = time.Now().Add(HEARTBEAT_EXPIRY)
		} else {
			worker.Delete(true)
		}
	case mdapi.MDPW_DISCONNECT:
		worker.Delete(false)
	default:
		log.Printf("E: invalid input message %q\n", msg)
	}
}
156465
func (c *Client) dispatchPMessage(message redis.PMessage) {
	if listeners, exists := c.subscribed[message.Pattern]; exists {
		for _, l := range listeners {
			l.PMessages <- message
		}
	}
}
132793
func (r *sentReq) deliver(peer distPeer, msg *Msg) error {
	r.lock.Lock()
	defer r.lock.Unlock()

	s, ok := r.sentTo[peer]
	if !ok || s.delivered {
		return errResp(ErrUnexpectedResponse, "reqID = %v", msg.ReqID)
	}
	valid := r.validate(peer, msg) == nil
	r.sentTo[peer] = sentReqToPeer{true, s.valid}
	s.valid <- valid
	if !valid {
		return errResp(ErrInvalidResponse, "reqID = %v", msg.ReqID)
	}
	return nil
}
1095119
func (p *Peer) HandleMsg(ctx context.Context, msg interface{}) error {
	select {
	case <-p.streamer.quit:
		log.Trace("message received after the streamer is closed", "peer", p.ID())
		// return without an error since streamer is closed and
		// no messages should be handled as other subcomponents like
		// storage leveldb may be closed
		return nil
	default:
	}

	switch msg := msg.(type) {

	case *SubscribeMsg:
		return p.handleSubscribeMsg(ctx, msg)

	case *SubscribeErrorMsg:
		return p.handleSubscribeErrorMsg(msg)

	case *UnsubscribeMsg:
		return p.handleUnsubscribeMsg(msg)

	case *OfferedHashesMsg:
		return p.handleOfferedHashesMsg(ctx, msg)

	case *TakeoverProofMsg:
		return p.handleTakeoverProofMsg(ctx, msg)

	case *WantedHashesMsg:
		return p.handleWantedHashesMsg(ctx, msg)

	case *ChunkDeliveryMsgRetrieval:
		// handling chunk delivery is the same for retrieval and syncing, so let's cast the msg
		return p.streamer.delivery.handleChunkDeliveryMsg(ctx, p, ((*ChunkDeliveryMsg)(msg)))

	case *ChunkDeliveryMsgSyncing:
		// handling chunk delivery is the same for retrieval and syncing, so let's cast the msg
		return p.streamer.delivery.handleChunkDeliveryMsg(ctx, p, ((*ChunkDeliveryMsg)(msg)))

	case *RetrieveRequestMsg:
		return p.streamer.delivery.handleRetrieveRequestMsg(ctx, p, msg)

	case *RequestSubscriptionMsg:
		return p.handleRequestSubscription(ctx, msg)

	case *QuitMsg:
		return p.handleQuitMsg(msg)

	default:
		return fmt.Errorf("unknown message type: %T", msg)
	}
}
604122
func Msg(logger log.Logger, message string, keyvals ...interface{}) error {
	prepended := structure.CopyPrepend(keyvals, structure.MessageKey, message)
	return logger.Log(prepended...)
}
354372
func (r *Request) EncodeMsg(w *msgp.Writer) error {
	var bm msgp.Encodable

	if r.Body != nil {
		var ok bool
		bm, ok = r.Body.(msgp.Encodable)
		if !ok {
			return ErrNotEncodable
		}
	}

	var err error

	if err = w.WriteArrayHeader(RequestPackSize); err != nil {
		return err
	}

	if err = w.WriteInt(RequestType); err != nil {
		return err
	}

	if err = w.WriteString(r.ID); err != nil {
		return err
	}

	if err = w.WriteString(r.Method); err != nil {
		return err
	}

	// No body to encode. Write a zero-length params array.
	if bm == nil {
		return w.WriteArrayHeader(0)
	}

	// 1-item array containing the body.
	if err = w.WriteArrayHeader(1); err != nil {
		return err
	}

	return msgp.Encode(w, bm)
}
803963
func (z interval16) MarshalMsg(b []byte) (o []byte, err error) {
	o = msgp.Require(b, z.Msgsize())
	// map header, size 2
	// string "start"
	o = append(o, 0x82, 0xa5, 0x73, 0x74, 0x61, 0x72, 0x74)
	o = msgp.AppendUint16(o, z.start)
	// string "last"
	o = append(o, 0xa4, 0x6c, 0x61, 0x73, 0x74)
	o = msgp.AppendUint16(o, z.last())
	return
}
1086799
func (z *IndexFindResp) MarshalMsg(b []byte) (o []byte, err error) {
	o = msgp.Require(b, z.Msgsize())
	// map header, size 1
	// string "Nodes"
	o = append(o, 0x81, 0xa5, 0x4e, 0x6f, 0x64, 0x65, 0x73)
	o = msgp.AppendMapHeader(o, uint32(len(z.Nodes)))
	for za0001, za0002 := range z.Nodes {
		o = msgp.AppendString(o, za0001)
		o = msgp.AppendArrayHeader(o, uint32(len(za0002)))
		for za0003 := range za0002 {
			o, err = za0002[za0003].MarshalMsg(o)
			if err != nil {
				err = msgp.WrapError(err, "Nodes", za0001, za0003)
				return
			}
		}
	}
	return
}
873779
func (d *deDupedAnnouncements) addMsg(message networkMsg) {
	// Depending on the message type (channel announcement, channel update,
	// or node announcement), the message is added to the corresponding map
	// in deDupedAnnouncements. Because each identifying key can have at
	// most one value, the announcements are de-duplicated, with newer ones
	// replacing older ones.
	switch msg := message.msg.(type) {

	// Channel announcements are identified by the short channel id field.
	case *lnwire.ChannelAnnouncement:
		deDupKey := msg.ShortChannelID
		sender := route.NewVertex(message.source)

		mws, ok := d.channelAnnouncements[deDupKey]
		if !ok {
			mws = msgWithSenders{
				msg:     msg,
				senders: make(map[route.Vertex]struct{}),
			}
			mws.senders[sender] = struct{}{}

			d.channelAnnouncements[deDupKey] = mws

			return
		}

		mws.msg = msg
		mws.senders[sender] = struct{}{}
		d.channelAnnouncements[deDupKey] = mws

	// Channel updates are identified by the (short channel id,
	// channelflags) tuple.
	case *lnwire.ChannelUpdate:
		sender := route.NewVertex(message.source)
		deDupKey := channelUpdateID{
			msg.ShortChannelID,
			msg.ChannelFlags,
		}

		oldTimestamp := uint32(0)
		mws, ok := d.channelUpdates[deDupKey]
		if ok {
			// If we already have seen this message, record its
			// timestamp.
			oldTimestamp = mws.msg.(*lnwire.ChannelUpdate).Timestamp
		}

		// If we already had this message with a strictly newer
		// timestamp, then we'll just discard the message we got.
		if oldTimestamp > msg.Timestamp {
			return
		}

		// If the message we just got is newer than what we previously
		// have seen, or this is the first time we see it, then we'll
		// add it to our map of announcements.
		if oldTimestamp < msg.Timestamp {
			mws = msgWithSenders{
				msg:     msg,
				senders: make(map[route.Vertex]struct{}),
			}

			// We'll mark the sender of the message in the
			// senders map.
			mws.senders[sender] = struct{}{}

			d.channelUpdates[deDupKey] = mws

			return
		}

		// Lastly, if we had seen this exact message from before, with
		// the same timestamp, we'll add the sender to the map of
		// senders, such that we can skip sending this message back in
		// the next batch.
		mws.msg = msg
		mws.senders[sender] = struct{}{}
		d.channelUpdates[deDupKey] = mws

	// Node announcements are identified by the Vertex field.  Use the
	// NodeID to create the corresponding Vertex.
	case *lnwire.NodeAnnouncement:
		sender := route.NewVertex(message.source)
		deDupKey := route.Vertex(msg.NodeID)

		// We do the same for node announcements as we did for channel
		// updates, as they also carry a timestamp.
		oldTimestamp := uint32(0)
		mws, ok := d.nodeAnnouncements[deDupKey]
		if ok {
			oldTimestamp = mws.msg.(*lnwire.NodeAnnouncement).Timestamp
		}

		// Discard the message if it's old.
		if oldTimestamp > msg.Timestamp {
			return
		}

		// Replace if it's newer.
		if oldTimestamp < msg.Timestamp {
			mws = msgWithSenders{
				msg:     msg,
				senders: make(map[route.Vertex]struct{}),
			}

			mws.senders[sender] = struct{}{}

			d.nodeAnnouncements[deDupKey] = mws

			return
		}

		// Add to senders map if it's the same as we had.
		mws.msg = msg
		mws.senders[sender] = struct{}{}
		d.nodeAnnouncements[deDupKey] = mws
	}
}
571320
func (b *Builder) Msg(m *dns.Msg) (err error) {
	if b.Full {
		err = b.Pack(m)
	}
	return
}
154176
func (s *StanServer) processMsg(c *channel) {
	ss := c.ss

	// Since we iterate through them all.
	ss.RLock()
	// Walk the plain subscribers and deliver to each one
	for _, sub := range ss.psubs {
		s.sendAvailableMessages(c, sub)
	}

	// Check the queue subscribers
	for _, qs := range ss.qsubs {
		s.sendAvailableMessagesToQueue(c, qs)
	}
	ss.RUnlock()
}
929426
func (ch *Channel) nextPacketMsg() PacketMsg {
	packet := PacketMsg{}
	packet.ChannelID = byte(ch.desc.ID)
	maxSize := ch.maxPacketMsgPayloadSize
	packet.Bytes = ch.sending[:cmn.MinInt(maxSize, len(ch.sending))]
	if len(ch.sending) <= maxSize {
		packet.EOF = byte(0x01)
		ch.sending = nil
		atomic.AddInt32(&ch.sendQueueSize, -1) // decrement sendQueueSize
	} else {
		packet.EOF = byte(0x00)
		ch.sending = ch.sending[cmn.MinInt(maxSize, len(ch.sending)):]
	}
	return packet
}
1026346
func (h *handler) handleMsg(msg *jsonrpcMessage) {
	if ok := h.handleImmediate(msg); ok {
		return
	}
	h.startCallProc(func(cp *callProc) {
		answer := h.handleCallMsg(cp, msg)
		h.addSubscriptions(cp.notifiers)
		if answer != nil {
			h.conn.Write(cp.ctx, answer)
		}
		for _, n := range cp.notifiers {
			n.activate()
		}
	})
}
173736
func (ev *msgEventer) ReadMsg() (Msg, error) {
	msg, err := ev.MsgReadWriter.ReadMsg()
	if err != nil {
		return msg, err
	}
	ev.feed.Send(&PeerEvent{
		Type:     PeerEventTypeMsgRecv,
		Peer:     ev.peerID,
		Protocol: ev.Protocol,
		MsgCode:  &msg.Code,
		MsgSize:  &msg.Size,
	})
	return msg, nil
}
839474
func (z *IndexFindResp) EncodeMsg(en *msgp.Writer) (err error) {
	// map header, size 1
	// write "Nodes"
	err = en.Append(0x81, 0xa5, 0x4e, 0x6f, 0x64, 0x65, 0x73)
	if err != nil {
		return
	}
	err = en.WriteMapHeader(uint32(len(z.Nodes)))
	if err != nil {
		err = msgp.WrapError(err, "Nodes")
		return
	}
	for za0001, za0002 := range z.Nodes {
		err = en.WriteString(za0001)
		if err != nil {
			err = msgp.WrapError(err, "Nodes")
			return
		}
		err = en.WriteArrayHeader(uint32(len(za0002)))
		if err != nil {
			err = msgp.WrapError(err, "Nodes", za0001)
			return
		}
		for za0003 := range za0002 {
			err = za0002[za0003].EncodeMsg(en)
			if err != nil {
				err = msgp.WrapError(err, "Nodes", za0001, za0003)
				return
			}
		}
	}
	return
}