Function topicInit

server/init_topic.go:21–135

topicInit reads an existing topic from database or creates a new topic

(t *Topic, join *ClientComMessage, h *Hub)

// topicInit reads an existing topic from database or creates a new topic
func topicInit(t *Topic, join *ClientComMessage, h *Hub) {
	var subscribeReqIssued bool
	defer func() {
		if !subscribeReqIssued && join.Sub != nil && join.sess.inflightReqs != nil {
			// If it was a client initiated subscribe request and we failed it.
			join.sess.inflightReqs.Done()
		}
	}()

	timestamp := types.TimeNow()

	var err error
	switch {
	case t.xoriginal == "me":
		// Request to load a 'me' topic. The topic always exists, the subscription is never new.
		err = initTopicMe(t, join)
	case t.xoriginal == "fnd":
		// Request to load a 'find' topic. The topic always exists, the subscription is never new.
		err = initTopicFnd(t, join)
	case strings.HasPrefix(t.xoriginal, "usr") || strings.HasPrefix(t.xoriginal, "p2p"):
		// Request to load an existing or create a new p2p topic, then attach to it.
		err = initTopicP2P(t, join)
	case strings.HasPrefix(t.xoriginal, "new"):
		// Processing request to create a new group topic.
		err = initTopicNewGrp(t, join, false)
	case strings.HasPrefix(t.xoriginal, "nch"):
		// Processing request to create a new channel.
		err = initTopicNewGrp(t, join, true)
	case strings.HasPrefix(t.xoriginal, "grp") || strings.HasPrefix(t.xoriginal, "chn"):
		// Load existing group topic (or channel).
		err = initTopicGrp(t)
	case t.xoriginal == "sys":
		// Initialize system topic.
		err = initTopicSys(t)
	case t.xoriginal == "slf":
		// Initialize self (notes and saved messages) topic.
		err = initTopicSlf(t, join)
	default:
		// Unrecognized topic name
		err = types.ErrTopicNotFound
	}

	// Failed to create or load the topic.
	if err != nil {
		// Remove topic from cache to prevent hub from forwarding more messages to it.
		h.topicDel(join.RcptTo)

		logs.Err.Println("init_topic: failed to load or create topic:", join.RcptTo, err)
		join.sess.queueOut(decodeStoreErrorExplicitTs(err, join.Id, t.xoriginal, timestamp, join.Timestamp, nil))

		// Re-queue pending requests to join the topic.
		for len(t.reg) > 0 {
			h.join <- (<-t.reg)
		}

		// Reject all other pending requests
		for len(t.clientMsg) > 0 {
			msg := <-t.clientMsg