Method handleSubChanges

db/blip_handler.go:266–396

CHANGES Received a "subChanges" subscription request

(rq *blip.Message)

// Received a "subChanges" subscription request
func (bh *blipHandler) handleSubChanges(rq *blip.Message) error {
	latestSeq := func() (SequenceID, error) {
		seq, err := bh.collection.LastSequence(bh.loggingCtx)
		return SequenceID{Seq: seq}, err
	}
	subChangesParams, err := NewSubChangesParams(bh.loggingCtx, rq, latestSeq, bh.db.Options.ChangesRequestPlus)
	if err != nil {
		return base.HTTPErrorf(http.StatusBadRequest, "Invalid subChanges parameters")
	}

	// Ensure that only _one_ subChanges subscription can be open on this blip connection at any given time.  SG #3222.
	collectionCtx := bh.collectionCtx
	collectionCtx.changesCtxLock.Lock()
	defer collectionCtx.changesCtxLock.Unlock()
	if !collectionCtx.activeSubChanges.CASRetry(false, true) {
		collectionStr := "default collection"
		if bh.collectionIdx != nil {
			collectionStr = fmt.Sprintf("collection %d", *bh.collectionIdx)
		}
		return fmt.Errorf("blipHandler for %s already has an outstanding subChanges. Cannot open another one", collectionStr)
	}

	// Create ctx if it has been cancelled
	if collectionCtx.changesCtx.Err() != nil {
		collectionCtx.changesCtx, collectionCtx.changesCtxCancel = context.WithCancel(bh.loggingCtx)
	}

	if len(subChangesParams.docIDs()) > 0 && subChangesParams.continuous() {
		return base.HTTPErrorf(http.StatusBadRequest, "DocIDs filter not supported for continuous subChanges")
	}

	bh.logEndpointEntry(rq.Profile(), subChangesParams.String())

	var channels base.Set
	if filter := subChangesParams.filter(); filter == base.ByChannelFilter {
		var err error

		channels, err = subChangesParams.channelsExpandedSet()
		if err != nil {
			return base.HTTPErrorf(http.StatusBadRequest, "%s", err)
		} else if len(channels) == 0 {
			return base.HTTPErrorf(http.StatusBadRequest, "Empty channel list")
		}
	} else if filter != "" {
		return base.HTTPErrorf(http.StatusBadRequest, "Unknown filter; try sync_gateway/bychannel")
	}

	collectionCtx.channels = channels

	clientType := clientTypeCBL2
	if rq.Properties["client_sgr2"] == trueProperty {
		clientType = clientTypeSGR2
	}

	continuous := subChangesParams.continuous()

	requestPlusSeq := uint64(0)
	// If non-continuous, check whether requestPlus handling is set for request or via database config