Method AddToCache

db/channel_cache.go:200–256

Adds an entry to the appropriate channels' caches, returning the affected channels. lateSequence flag indicates whether it was a change arriving out of sequence

(ctx context.Context, change *LogEntry)

// Adds an entry to the appropriate channels' caches, returning the affected channels.  lateSequence
// flag indicates whether it was a change arriving out of sequence
func (c *channelCacheImpl) AddToCache(ctx context.Context, change *LogEntry) []channels.ID {

	ch := change.Channels
	change.Channels = nil // not needed anymore, so free some memory

	// updatedChannels tracks the set of channels that should be notified of the change.  This includes
	// the change's active channels, as well as any channel removals for the active revision.
	updatedChannels := make([]channels.ID, 0, len(ch)+1) // +1 for the star channel

	// If it's a late sequence, we want to add to all channel late queues within a single write lock,
	// to avoid a changes feed seeing the same late sequence in different iteration loops (and sending
	// twice)
	if change.Skipped {
		c.lateSeqLock.Lock()
		defer c.lateSeqLock.Unlock()
	}

	// Need to acquire the validFromLock prior to checking for active channel caches, to ensure that
	// any new caches that are added between the check for c.GetActiveChannelCache and the update of
	// c.highCacheSequence are initialized with the correct validFrom.
	var explicitStarChannel bool
	c.validFromLock.Lock()
	for channelName, removal := range ch {
		if removal == nil || removal.Seq == change.Sequence {
			// If the document has been explicitly added to the star channel by the sync function, don't need to recheck below
			if channelName == channels.UserStarChannel {
				explicitStarChannel = true
			}
			channelID := channels.NewID(channelName, change.CollectionID)
			channelCache, ok := c.getActiveChannelCache(ctx, channelID)
			if ok {
				channelCache.addToCache(ctx, change, removal != nil)
				if change.Skipped {
					channelCache.AddLateSequence(change)
				}
			}
			// Need to notify even if channel isn't active, for case where number of connected changes channels exceeds cache capacity
			updatedChannels = append(updatedChannels, channelID)
		}
	}

	if EnableStarChannelLog && !explicitStarChannel {
		starChannelID := channels.NewID(channels.UserStarChannel, change.CollectionID)
		channelCache, ok := c.getActiveChannelCache(ctx, starChannelID)
		if ok {
			channelCache.addToCache(ctx, change, false)
			if change.Skipped {
				channelCache.AddLateSequence(change)
			}
		}
		updatedChannels = append(updatedChannels, starChannelID)
	}

	c.updateHighCacheSequence(change.Sequence)
	c.validFromLock.Unlock()
	return updatedChannels
}