Method run

server/hub.go:151–344

()

}

func (h *Hub) run() {
	for {
		select {
		case join := <-h.join:
			// Handle a subscription request:
			// 1. Init topic
			// 1.1 If a new topic is requested, create it
			// 1.2 If a new subscription to an existing topic is requested:
			// 1.2.1 check if topic is already loaded
			// 1.2.2 if not, load it
			// 1.2.3 if it cannot be loaded (not found), fail
			// 2. Check access rights and reject, if appropriate
			// 3. Attach session to the topic
			// Is the topic already loaded?
			t := h.topicGet(join.RcptTo)
			if t == nil {
				// Topic does not exist or not loaded.
				t = &Topic{
					name:      join.RcptTo,
					xoriginal: join.Original,
					// Indicates a proxy topic.
					isProxy:   globals.cluster.isRemoteTopic(join.RcptTo),
					sessions:  make(map[*Session]perSessionData),
					clientMsg: make(chan *ClientComMessage, 192),
					serverMsg: make(chan *ServerComMessage, 64),
					reg:       make(chan *ClientComMessage, 256),
					unreg:     make(chan *ClientComMessage, 256),
					meta:      make(chan *ClientComMessage, 64),
					perUser:   make(map[types.Uid]perUserData),
					exit:      make(chan *shutDown, 1),
				}
				if globals.cluster != nil {
					if t.isProxy {
						t.proxy = make(chan *ClusterResp, 128)
						t.masterNode = globals.cluster.ring.Get(t.name)
					} else {
						// It's a master topic. Make a channel for handling
						// direct messages from the proxy.
						t.master = make(chan *ClusterSessUpdate, 8)
					}
				}
				// Topic is created in suspended state because it's not yet configured.
				t.markPaused(true)
				// Save topic now to prevent race condition.
				h.topicPut(join.RcptTo, t)

				// Configure the topic.
				go topicInit(t, join, h)
			} else {
				// Topic found.
				if t.isInactive() {
					// Topic is either not ready or being deleted.
					if join.sess.inflightReqs != nil {
						join.sess.inflightReqs.Done()
					}
					join.sess.queueOut(ErrLockedReply(join, join.Timestamp))
					continue
				}