Function usersRegisterTopic

server/user.go:871–910

Account users as members of an active topic. Used for cache management. In case of a cluster this method is called only when the topic is local: globals.cluster.isRemoteTopic(t.name) == false

(t *Topic, add bool)

// In case of a cluster this method is called only when the topic is local:
// globals.cluster.isRemoteTopic(t.name) == false
func usersRegisterTopic(t *Topic, add bool) {
	if globals.usersUpdate == nil {
		return
	}

	if t.cat == types.TopicCatFnd || t.cat == types.TopicCatMe {
		// Ignoring me and fnd topics.
		return
	}

	local := &UserCacheReq{Inc: add}

	// In case of a cluster UIDs could be local and remote. Process local UIDs locally,
	// send remote UIDs to other cluster nodes for processing. The UIDs may have to be
	// sent to multiple nodes.
	remote := &UserCacheReq{Inc: add}
	for uid, pud := range t.perUser {
		if pud.isChan {
			// Skip channel subscribers.
			continue
		}
		if globals.cluster.isRemoteTopic(uid.UserId()) {
			remote.UserIdList = append(remote.UserIdList, uid)
		} else {
			local.UserIdList = append(local.UserIdList, uid)
		}
	}

	if len(remote.UserIdList) > 0 {
		globals.cluster.routeUserReq(remote)
	}

	if len(local.UserIdList) > 0 {
		select {
		case globals.usersUpdate <- local:
		default:
			logs.Err.Println("User cache: globals.usersUpdate queue full: ", len(globals.usersUpdate))
		}
	}
}

// usersRequestFromCluster handles requests which came from other cluser nodes.
func usersRequestFromCluster(req *UserCacheReq) {