Method SimpleMultiChangesFeed

db/changes.go:709–1205

Returns the (ordered) union of all of the changes made to multiple channels.

(ctx context.Context, chans base.Set, options ChangesOptions)

// Returns the (ordered) union of all of the changes made to multiple channels.
func (col *DatabaseCollectionWithUser) SimpleMultiChangesFeed(ctx context.Context, chans base.Set, options ChangesOptions) (<-chan *ChangeEntry, error) {

	to := ""
	if col.user != nil && col.user.Name() != "" {
		to = fmt.Sprintf("  (to %s)", col.user.Name())
	} else {
		to = "  (to ADMIN)"
	}

	base.InfofCtx(ctx, base.KeyChanges, "MultiChangesFeed(channels: %s, options: %s) ... %s", base.UD(chans), options, base.UD(to))
	output := make(chan *ChangeEntry, 50)

	collectionID := col.GetCollectionID()
	go func() {

		defer func() {
			if panicked := recover(); panicked != nil {
				base.WarnfCtx(ctx, "[%s] Unexpected panic sending changes - terminating changes: \n %s", panicked, debug.Stack())
			} else {
				base.InfofCtx(ctx, base.KeyChanges, "MultiChangesFeed done %s", base.UD(to))
			}
			close(output)
		}()

		var changeWaiter *ChangeWaiter
		var lowSequence uint64
		var currentCachedSequence uint64 // The highest contiguous sequence buffered over the caching feed
		var lateSequenceFeeds map[channels.ID]*lateSequenceFeed
		var useLateSequenceFeeds bool            // LateSequence feeds are only used for continuous, or one-shot where options.RequestPlusSeq > currentCachedSequence
		var userCounter uint64                   // Wait counter used to identify changes to the user document
		var changedChannels channels.ChangedKeys // Tracks channels added/removed to the user during changes processing.
		var userChanged bool                     // Whether the user document has changed in a given iteration loop
		var deferredBackfill bool                // Whether there's a backfill identified in the user doc that's deferred while the SG cache catches up

		// Retrieve the current max cached sequence - ensures there isn't a race between the subsequent channel cache queries
		currentCachedSequence = col.changeCache().getChannelCache().GetHighCacheSequence()

		// If changes feed requires more than one ChangesLoop iteration, initialize changeWaiter
		if options.Wait || options.RequestPlusSeq > currentCachedSequence {
			trackUnusedSequences := options.RequestPlusSeq > 0
			changeWaiter = col.startChangeWaiter(trackUnusedSequences) // Waiter is updated with the actual channel set (post-user reload) at the start of the outer changes loop
			userCounter = changeWaiter.CurrentUserCount()
			// Reload user to pick up user changes that happened between auth and the change waiter
			// initialization.  Without this, notification for user doc changes in that window (a) won't be
			// included in the initial changes loop iteration, and (b) won't wake up the ChangeWaiter.
			if col.user != nil {
				previousRoles := col.user.RoleNames()
				if err := col.ReloadUser(ctx); err != nil {
					base.WarnfCtx(ctx, "Error reloading user during changes initialization %q: %v", base.UD(col.user.Name()), err)
					change := makeErrorEntry("User not found during reload - terminating changes feed")
					output <- &change
					return
				}
				changedRoles := col.user.RoleNames().CompareKeys(previousRoles)
				if len(changedRoles) > 0 {
					changeWaiter.RefreshUserKeys(col.User(), col.dbCtx.MetadataKeys)
				}
			}