Method handleChanges

db/blip_handler.go:666–814

Handles a "changes" request, i.e. a set of changes pushed by the client

(rq *blip.Message)

// Handles a "changes" request, i.e. a set of changes pushed by the client
func (bh *blipHandler) handleChanges(rq *blip.Message) error {
	var ignoreNoConflicts bool
	if val := rq.Properties[ChangesMessageIgnoreNoConflicts]; val != "" {
		ignoreNoConflicts = val == trueProperty
	}

	if !ignoreNoConflicts && !bh.collection.AllowConflicts() {
		return ErrUseProposeChanges
	}

	var changeList [][]interface{}
	if err := rq.ReadJSONBody(&changeList); err != nil {
		base.WarnfCtx(bh.loggingCtx, "Handle changes got error: %v", err)
		return err
	}

	collectionCtx := bh.collectionCtx
	bh.logEndpointEntry(rq.Profile(), fmt.Sprintf("#Changes:%d", len(changeList)))
	if len(changeList) == 0 {
		// An empty changeList is sent when a one-shot replication sends its final changes
		// message, or a continuous replication catches up *for the first time*.
		// Note that this doesn't mean that rev messages associated with previous changes
		// messages have been fully processed
		if collectionCtx.emptyChangesMessageCallback != nil {
			collectionCtx.emptyChangesMessageCallback()
		}
		return nil
	}
	output := bytes.NewBuffer(make([]byte, 0, 100*len(changeList)))
	output.Write([]byte("["))
	jsonOutput := base.JSONEncoder(output)
	nWritten := 0
	nRequested := 0

	// Include changes messages w/ proposeChanges stats, although CBL should only be using proposeChanges
	startTime := time.Now()
	bh.replicationStats.HandleChangesCount.Add(int64(len(changeList)))
	defer func() {
		bh.replicationStats.HandleChangesTime.Add(time.Since(startTime).Nanoseconds())
	}()

	// DocID+RevID -> SeqNo
	expectedSeqs := make(map[IDAndRev]SequenceID, 0)
	alreadyKnownSeqs := make([]SequenceID, 0)

	versionVectorProtocol := bh.useHLV()

	for _, change := range changeList {
		docID := change[1].(string)
		rev := change[2].(string)
		var missing, possible []string

		if !versionVectorProtocol {
			missing, possible = bh.collection.RevDiff(bh.loggingCtx, docID, []string{rev})
		} else {
			missing, possible = bh.collection.CheckChangeVersion(bh.loggingCtx, docID, rev)
		}