// Queues a message to be sent to channel.
778819
func (c *MConnection) Send(chID byte, msgBytes []byte) bool {
	if !c.IsRunning() {
		return false
	}

	c.Logger.Debug("Send", "channel", chID, "conn", c, "msgBytes", fmt.Sprintf("%X", msgBytes))

	// Send message to channel.
	channel, ok := c.channelsIdx[chID]
	if !ok {
		c.Logger.Error(fmt.Sprintf("Cannot send bytes, unknown channel %X", chID))
		return false
	}

	success := channel.sendBytes(msgBytes)
	if success {
		// Wake up sendRoutine if necessary
		select {
		case c.send <- struct{}{}:
		default:
		}
	} else {
		c.Logger.Debug("Send failed", "channel", chID, "conn", c, "msgBytes", fmt.Sprintf("%X", msgBytes))
	}
	return success
}
305176
func (c *EncodedConn) QueueSubscribe(subject, queue string, cb Handler) (*Subscription, error) {
	return c.subscribe(subject, queue, cb)
}
1236076
func (router *Metadata) Enqueue(msg *core.Message) error {
	metadata := msg.TryGetMetadata()
	if metadata == nil {
		return router.Broadcast.Enqueue(msg)
	}

	streamName := metadata.GetValueString(router.key)
	targetID := core.GetStreamID(streamName)
	if targetID == core.InvalidStreamID {
		return router.Broadcast.Enqueue(msg)
	}

	targetRouter := core.StreamRegistry.GetRouterOrFallback(targetID)

	msg.SetStreamID(targetID)
	return core.Route(msg, targetRouter)
}
152415
func (e RTMEvent) Message() (*RTMMessage, error) {
	var msg RTMMessage
	if err := json.Unmarshal(e.RawMessage, &msg); err != nil {
		return nil, err
	}

	return &msg, nil
}
651399
func (r *messageRing) Enqueue(m *Message) error {
	mSize := int64(len(m.Line))

	r.mu.Lock()
	if r.closed {
		r.mu.Unlock()
		return errClosed
	}
	if mSize+r.sizeBytes > r.maxBytes && len(r.queue) > 0 {
		r.wait.Signal()
		r.mu.Unlock()
		return nil
	}

	r.queue = append(r.queue, m)
	r.sizeBytes += mSize
	r.wait.Signal()
	r.mu.Unlock()
	return nil
}
355418
func (c *Client) Message(level string, msg string) {
	c.MessageWithExtras(level, msg, noExtras)
}
1047585
func (l Players) BroadcastMessage(msg string) {
	l.RLock()
	for _, p := range l.list {
		if p.Ready {
			p.SendMessage(msg)
		}
	}
	l.RUnlock()
}
582641
func (c *Client) QueueMsg(msg *Message) error {
	if msg.Timestamp == nil {
		curTime := time.Now()
		msg.Timestamp = &curTime
	}

	c.msgChan <- msg
	return nil
}
1030632
func (c *Channel) SendMessage(message *Message) {
	c.lastEventID = message.id

	for c, open := range c.clients {
		if open {
			c.send <- message
		}
	}
}
705548
func (c *channel) responseMessageReceived() error {
	if c.direction == channelInbound {
		return errors.New("ssh: channel response message received on inbound channel")
	}
	if c.decided {
		return errors.New("ssh: duplicate response received for channel")
	}
	c.decided = true
	return nil
}
126578
func (ch *Channel) dispatch(msg message) {
	switch m := msg.(type) {
	case *channelClose:
		// lock before sending connection.close-ok
		// to avoid unexpected interleaving with basic.publish frames if
		// publishing is happening concurrently
		ch.m.Lock()
		ch.send(&channelCloseOk{})
		ch.m.Unlock()
		ch.connection.closeChannel(ch, newError(m.ReplyCode, m.ReplyText))

	case *channelFlow:
		ch.notifyM.RLock()
		for _, c := range ch.flows {
			c <- m.Active
		}
		ch.notifyM.RUnlock()
		ch.send(&channelFlowOk{Active: m.Active})

	case *basicCancel:
		ch.notifyM.RLock()
		for _, c := range ch.cancels {
			c <- m.ConsumerTag
		}
		ch.notifyM.RUnlock()
		ch.consumers.cancel(m.ConsumerTag)

	case *basicReturn:
		ret := newReturn(*m)
		ch.notifyM.RLock()
		for _, c := range ch.returns {
			c <- *ret
		}
		ch.notifyM.RUnlock()

	case *basicAck:
		if ch.confirming {
			if m.Multiple {
				ch.confirms.Multiple(Confirmation{m.DeliveryTag, true})
			} else {
				ch.confirms.One(Confirmation{m.DeliveryTag, true})
			}
		}

	case *basicNack:
		if ch.confirming {
			if m.Multiple {
				ch.confirms.Multiple(Confirmation{m.DeliveryTag, false})
			} else {
				ch.confirms.One(Confirmation{m.DeliveryTag, false})
			}
		}

	case *basicDeliver:
		ch.consumers.send(m.ConsumerTag, newDelivery(ch, m))
		// TODO log failed consumer and close channel, this can happen when
		// deliveries are in flight and a no-wait cancel has happened

	default:
		ch.rpc <- msg
	}
}
1112240
func (c *Client) sendToChannel(keyed keyedData) {
	c.Lock()
	defer c.Unlock()

	for key, channel := range c.watchers {
		if strings.Contains(keyed.Key, key) {
			channel <- keyed
			return
		}
	}
}
198514
func (sm *SyncManager) QueueInv(inv *wire.MsgInv, peer *peerpkg.Peer) {
	// No channel handling here because peers do not need to block on inv
	// messages.
	if atomic.LoadInt32(&sm.shutdown) != 0 {
		return
	}

	sm.msgChan <- &invMsg{inv: inv, peer: peer}
}
809036
func (m *Memberlist) queueBroadcast(node string, msg []byte, notify chan struct{}) {
	b := &memberlistBroadcast{node, msg, notify}
	m.broadcasts.QueueBroadcast(b)
}
920758
func (s *ServerStream) SendMessage(ctx context.Context, msg *StreamMessage) error {
	return s.stream.SendMessage(ctx, msg)
}
772962
func (p *Player) BroadcastMessage(toList map[string]*Player, message string) {
	for _, to := range toList {
		// Send message only to other ready players
		if to.Ready && to.Name != p.Name {
			to.SendMessage(message)
		}
	}
}
756339
func (s *StanServer) sendAvailableMessages(c *channel, sub *subState) {
	sub.Lock()
	for nextSeq := sub.LastSent + 1; !sub.stalled; nextSeq++ {
		nextMsg := s.getNextMsg(c, &nextSeq, &sub.LastSent)
		if nextMsg == nil {
			break
		}
		if sent, sendMore := s.sendMsgToSub(sub, nextMsg, honorMaxInFlight); !sent || !sendMore {
			break
		}
	}
	sub.Unlock()
}