Method addChannelCache

db/channel_cache.go:376–411

Adds a new channel to the channel cache. Locking seqLock is required here to prevent missed data in the following scenario: // 1. addChannelCache issued for channel A // 2. addChannelCache obtains stable sequence, seq=10 // 3. addToCache (from another goroutine) receives sequence 11

(ctx context.Context, channel channels.ID)

//	//  This scenario would result in sequence 11 missing from the cache.  Locking seqLock ensures that
//	//  step 3 blocks until step 4 is complete (and so sees the channel as active)
func (c *channelCacheImpl) addChannelCache(ctx context.Context, channel channels.ID) (*singleChannelCacheImpl, bool) {

	// Return nil if the cache at capacity.
	if c.channelCaches.Length() >= c.maxChannels {
		return nil, false
	}

	// Return nil if a queryHandler can't be obtained for the collectionID
	queryHandler, err := c.queryHandlerFactory(channel.CollectionID)
	if err != nil {
		return nil, false
	}

	c.validFromLock.Lock()

	// Everything after the current high sequence will be added to the cache via the feed
	validFrom := c.GetHighCacheSequence() + 1

	singleChannelCache :=
		newChannelCacheWithOptions(ctx, queryHandler, channel, validFrom, c.options, c.cacheStats)
	cacheValue, created, cacheSize := c.channelCaches.GetOrInsert(channel, singleChannelCache)
	c.validFromLock.Unlock()

	singleChannelCache = AsSingleChannelCache(ctx, cacheValue)

	if cacheSize > c.compactHighWatermark {
		c.startCacheCompaction(ctx)
	}

	if created {
		c.cacheStats.ChannelCacheNumChannels.Add(1)
		c.cacheStats.ChannelCacheChannelsAdded.Add(1)
	}

	return singleChannelCache, true
}

func (c *channelCacheImpl) getActiveChannelCache(ctx context.Context, channel channels.ID) (*singleChannelCacheImpl, bool) {