Method subscribe

server/session.go:617–647

Request to subscribe to a topic.

(msg *ClientComMessage)

// Request to subscribe to a topic.
func (s *Session) subscribe(msg *ClientComMessage) {
	if strings.HasPrefix(msg.Original, "new") || strings.HasPrefix(msg.Original, "nch") {
		// Request to create a new group/channel topic.
		// If we are in a cluster, make sure the new topic belongs to the current node.
		msg.RcptTo = globals.cluster.genLocalTopicName()
	} else {
		var resp *ServerComMessage
		msg.RcptTo, resp = s.expandTopicName(msg)
		if resp != nil {
			s.queueOut(resp)
			return
		}
	}

	s.inflightReqs.Add(1)
	// Session can subscribe to topic on behalf of a single user at a time.
	if sub := s.getSub(msg.RcptTo); sub != nil {
		s.queueOut(InfoAlreadySubscribed(msg.Id, msg.Original, msg.Timestamp))
		s.inflightReqs.Done()
	} else {
		select {
		case globals.hub.join <- msg:
		default:
			// Reply with a 503 to the user.
			s.queueOut(ErrServiceUnavailableReply(msg, msg.Timestamp))
			s.inflightReqs.Done()
			logs.Err.Println("s.subscribe: hub.join queue full, topic ", msg.RcptTo, s.sid)
		}
		// Hub will send Ctrl success/failure packets back to session
	}
}

// Leave/Unsubscribe a topic
func (s *Session) leave(msg *ClientComMessage) {