Method TopicMaster

server/cluster.go:469–613

TopicMaster is a gRPC endpoint which receives requests sent by proxy topic to master topic.

(msg *ClusterReq, rejected *bool)

// TopicMaster is a gRPC endpoint which receives requests sent by proxy topic to master topic.
func (c *Cluster) TopicMaster(msg *ClusterReq, rejected *bool) error {
	*rejected = false

	node := c.nodes[msg.Node]
	if node == nil {
		logs.Warn.Println("cluster TopicMaster: request from an unknown node", msg.Node)
		return nil
	}

	// Master maintains one multiplexing session per proxy topic per node.
	// Except channel topics:
	// * one multiplexing session for channel subscriptions.
	// * one multiplexing session for group subscriptions.
	var msid string
	if msg.CliMsg != nil && types.IsChannel(msg.CliMsg.Original) {
		// If it's a channel request, use channel name.
		msid = msg.CliMsg.Original
	} else {
		msid = msg.RcptTo
	}
	// Append node name.
	msid += "-" + msg.Node
	msess := globals.sessionStore.Get(msid)

	if msg.Gone {
		// Proxy topic is gone. Tear down the local auxiliary session.
		// If it was the last session, master topic will shut down as well.
		node.stopMultiplexingSession(msess)

		if t := globals.hub.topicGet(msg.RcptTo); t != nil && t.isChan {
			// If it's a channel topic, also stop the "chnX-" local auxiliary session.
			msidChn := types.GrpToChn(t.name) + "-" + msg.Node
			node.stopMultiplexingSession(globals.sessionStore.Get(msidChn))
		}

		return nil
	}

	if msg.Signature != c.ring.Signature() {
		logs.Warn.Println("cluster TopicMaster: session signature mismatch", msg.RcptTo)
		*rejected = true
		return nil
	}

	// Create a new multiplexing session if needed.
	if msess == nil {
		// If the session is not found, create it.
		var count int
		msess, count = globals.sessionStore.NewSession(node, msid)
		node.lock.Lock()
		node.msess[msid] = struct{}{}
		node.lock.Unlock()

		logs.Info.Println("cluster: multiplexing session started", msid, count)
		msess.proxiedTopic = msg.RcptTo
	}

	// This is a local copy of a remote session.