Method sendBatchOfChanges

db/blip_handler.go:597–663

(sender *blip.Sender, changeArray [][]interface{}, ignoreNoConflicts bool)

}

func (bh *blipHandler) sendBatchOfChanges(sender *blip.Sender, changeArray [][]interface{}, ignoreNoConflicts bool) error {
	outrq := blip.NewRequest()
	outrq.SetProfile("changes")
	if ignoreNoConflicts {
		outrq.Properties[ChangesMessageIgnoreNoConflicts] = trueProperty
	}
	if bh.collectionIdx != nil {
		outrq.Properties[BlipCollection] = strconv.Itoa(*bh.collectionIdx)
	}
	err := outrq.SetJSONBody(changeArray)
	if err != nil {
		base.InfofCtx(bh.loggingCtx, base.KeyAll, "Error setting changes: %v", err)
	}

	if len(changeArray) > 0 {
		// Check for user updates before creating the db copy for handleChangesResponse
		if err := bh.refreshUser(); err != nil {
			return err
		}

		handleChangesResponseDbCollection, err := bh.copyDatabaseCollectionWithUser(bh.collectionIdx)
		if err != nil {
			return err
		}

		sendTime := time.Now()
		if !bh.sendBLIPMessage(sender, outrq) {
			return ErrClosedBLIPSender
		}

		bh.inFlightChangesThrottle <- struct{}{}
		atomic.AddInt64(&bh.changesPendingResponseCount, 1)

		bh.replicationStats.SendChangesCount.Add(int64(len(changeArray)))
		// Spawn a goroutine to await the client's response:
		go func(bh *blipHandler, sender *blip.Sender, response *blip.Message, changeArray [][]interface{}, sendTime time.Time, dbCollection *DatabaseCollectionWithUser) {
			if err := bh.handleChangesResponse(bh.loggingCtx, sender, response, changeArray, sendTime, dbCollection, bh.collectionIdx); err != nil && !errors.Is(err, ErrClosedBLIPSender) {
				base.WarnfCtx(bh.loggingCtx, "Error from bh.handleChangesResponse: %v", err)
				if bh.fatalErrorCallback != nil {
					bh.fatalErrorCallback(err)
				}
			}

			// Sent all of the revs for this changes batch, allow another changes batch to be sent.
			select {
			case <-bh.inFlightChangesThrottle:
			case <-bh.terminator:
			}

			atomic.AddInt64(&bh.changesPendingResponseCount, -1)
		}(bh, sender, outrq.Response(), changeArray, sendTime, handleChangesResponseDbCollection)
	} else {
		outrq.SetNoReply(true)
		if !bh.sendBLIPMessage(sender, outrq) {
			return ErrClosedBLIPSender
		}
	}