Method Init

db/change_cache.go:152–194

HOUSEKEEPING: Initializes a new changeCache. lastSequence is the last known database sequence assigned. notifyChangeFunc is an optional function that will be called to notify of channel changes. After calling Init(), you must call .Start() to start using the cache, otherwise it will be in a locked s

(ctx context.Context, dbContext *DatabaseContext, channelCache ChannelCache, notifyChange func(context.Context, channels.Set), options *CacheOptions, metaKeys *base.MetadataKeys)

// and callers will block on trying to obtain the lock.

func (c *changeCache) Init(ctx context.Context, dbContext *DatabaseContext, channelCache ChannelCache, notifyChange func(context.Context, channels.Set), options *CacheOptions, metaKeys *base.MetadataKeys) error {
	c.db = dbContext
	c.logCtx = ctx

	c.notifyChangeFunc = notifyChange
	c.receivedSeqs = make(map[uint64]struct{})
	c.terminator = make(chan bool)
	c.initTime = time.Now()
	c.skippedSeqs = NewSkippedSequenceSkiplist()
	c.lastAddPendingTime = time.Now().UnixNano()
	c.sgCfgPrefix = dbContext.MetadataKeys.SGCfgPrefix(c.db.Options.GroupID)
	c.metaKeys = metaKeys

	// init cache options
	if options != nil {
		c.options = *options
	} else {
		c.options = DefaultCacheOptions()
	}

	c.channelCache = channelCache

	base.InfofCtx(ctx, base.KeyCache, "Initializing changes cache for %s with options %+v", base.UD(c.db.Name), c.options)

	heap.Init(&c.pendingLogs)

	// background tasks that perform housekeeping duties on the cache
	bgt, err := NewBackgroundTask(ctx, "InsertPendingEntries", c.InsertPendingEntries, c.options.CachePendingSeqMaxWait/2, c.terminator)
	if err != nil {
		return err
	}
	c.backgroundTasks = append(c.backgroundTasks, bgt)

	bgt, err = NewBackgroundTask(ctx, "CleanSkippedSequenceQueue", c.CleanSkippedSequenceQueue, c.options.CacheSkippedSeqMaxWait/2, c.terminator)
	if err != nil {
		return err
	}
	c.backgroundTasks = append(c.backgroundTasks, bgt)

	// Lock the cache -- not usable until .Start() called.  This fixes the DCP startup race condition documented in SG #3558.
	c.lock.Lock()
	return nil
}

func (c *changeCache) Start(initialSequence uint64) error {