Method replySetSub

server/topic.go:2718–2764

replySetSub is a response to new subscription request or an update to a subscription {set.sub}: update topic metadata cache, save/update subs, reply to the caller as {ctrl} message, generate a presence notification, if appropriate.

(sess *Session, pkt *ClientComMessage, asChan bool)

// update topic metadata cache, save/update subs, reply to the caller as {ctrl} message,
// generate a presence notification, if appropriate.
func (t *Topic) replySetSub(sess *Session, pkt *ClientComMessage, asChan bool) error {
	now := types.TimeNow()

	asUid := types.ParseUserId(pkt.AsUser)
	set := pkt.Set

	var target types.Uid
	if target = types.ParseUserId(set.Sub.User); target.IsZero() && set.Sub.User != "" {
		// Invalid user ID
		sess.queueOut(ErrMalformedReply(pkt, now))
		return errors.New("invalid user id")
	}

	// if set.User is not set, request is for the current user
	if target.IsZero() {
		target = asUid
	}

	var err error
	var modeChanged *MsgAccessMode
	if target == asUid {
		// Request new subscription or modify own subscription
		modeChanged, err = t.thisUserSub(sess, pkt, asUid, asChan, set.Sub.Mode, nil)
	} else {
		// Request to approve/change someone's subscription
		modeChanged, err = t.anotherUserSub(sess, asUid, target, asChan, pkt)
	}
	if err != nil {
		return err
	}

	var resp *ServerComMessage
	if modeChanged != nil {
		// Report resulting access mode.
		params := map[string]any{"acs": modeChanged}
		if target != asUid {
			params["user"] = target.UserId()
		}
		resp = NoErrParamsReply(pkt, now, params)
	} else {
		resp = InfoNotModifiedReply(pkt, now)
	}

	sess.queueOut(resp)

	return nil
}

// replyGetData is a response to a get.data request - load a list of stored messages, send them to session as {data}
// response goes to a single session rather than all sessions in a topic