Function GenerateChanges

db/changes.go:1508–1656

Shell of the continuous changes feed -- calls out to a `send` function to deliver the change. This is called from BLIP connections as well as HTTP handlers, which is why this is not a method on `handler`.

(ctx context.Context, database *DatabaseCollectionWithUser, inChannels base.Set, options ChangesOptions, docIDFilter []string, send func([]*ChangeEntry) error)

// This is called from BLIP connections as well as HTTP handlers, which is why this is not a
// method on `handler`.
func GenerateChanges(ctx context.Context, database *DatabaseCollectionWithUser, inChannels base.Set, options ChangesOptions, docIDFilter []string, send func([]*ChangeEntry) error) (err error, forceClose bool) {
	// Set up heartbeat/timeout
	var timeoutInterval time.Duration
	var timer *time.Timer
	var heartbeat <-chan time.Time
	if options.HeartbeatMs > 0 {
		ticker := time.NewTicker(time.Duration(options.HeartbeatMs) * time.Millisecond)
		defer ticker.Stop()
		heartbeat = ticker.C
	} else if options.TimeoutMs > 0 {
		timeoutInterval = time.Duration(options.TimeoutMs) * time.Millisecond
		defer func() {
			if timer != nil {
				timer.Stop()
			}
		}()
	}

	if options.Continuous {
		options.Wait = true // we want the feed channel to wait for changes
	}

	if !options.Since.IsNonZero() {
		database.dbStats().CBLReplicationPull().NumPullReplSinceZero.Add(1)
	}

	var lastSeq SequenceID
	var feed <-chan *ChangeEntry
	var timeout <-chan time.Time
	var feedErr error

	// feedStarted identifies whether at least one MultiChangesFeed has been started.  Used to identify when a one-shot changes is done.
	feedStarted := false

loop:
	for {
		// If the feed has already been started once and closed, and this isn't a continuous
		// replication, we're done.
		if feedStarted && feed == nil && !options.Continuous {
			break loop
		}

		if feed == nil {
			// Refresh the feed of all current changes:
			if lastSeq.IsNonZero() { // start after end of last feed
				options.Since = lastSeq
			}
			if database.IsClosed() {
				forceClose = true
				break loop
			}
			if len(docIDFilter) > 0 {
				feed, feedErr = database.DocIDChangesFeed(ctx, inChannels, docIDFilter, options)
			} else {
				feed, feedErr = database.MultiChangesFeed(ctx, inChannels, options)
			}
			if feedErr != nil || feed == nil {
				return feedErr, forceClose