Function NewInMemoryAutoRefresh

flytestdlib/cache/in_memory_auto_refresh.go:120–164

NewInMemoryAutoRefresh creates a new InMemoryAutoRefresh

(
	name string,
	syncCb SyncFunc,
	syncRateLimiter workqueue.TypedRateLimiter[*Batch],
	resyncPeriod time.Duration,
	parallelizm uint,
	size uint,
	scope promutils.Scope,
	options ...Option,
)

// NewInMemoryAutoRefresh creates a new InMemoryAutoRefresh
func NewInMemoryAutoRefresh(
	name string,
	syncCb SyncFunc,
	syncRateLimiter workqueue.TypedRateLimiter[*Batch],
	resyncPeriod time.Duration,
	parallelizm uint,
	size uint,
	scope promutils.Scope,
	options ...Option,
) (*InMemoryAutoRefresh, error) {
	opts := defaultOptions()
	for _, option := range options {
		option(opts)
	}

	metrics := newMetrics(scope)
	// #nosec G115
	lruCache, err := lru.NewWithEvict(int(size), getEvictionFunction(metrics.Evictions))
	if err != nil {
		return nil, fmt.Errorf("creating LRU cache: %w", err)
	}

	cache := &InMemoryAutoRefresh{
		name:            name,
		metrics:         metrics,
		parallelizm:     parallelizm,
		createBatchesCb: opts.createBatchesCb,
		syncCb:          syncCb,
		lruMap:          lruCache,
		processing:      &sync.Map{},
		toDelete:        newSyncSet(),
		syncPeriod:      resyncPeriod,
		workqueue: workqueue.NewTypedRateLimitingQueueWithConfig(syncRateLimiter, workqueue.TypedRateLimitingQueueConfig[*Batch]{
			Name:  scope.CurrentScope(),
			Clock: opts.clock,
		}),
		clock:              opts.clock,
		syncCount:          atomic.NewInt32(0),
		enqueueCount:       atomic.NewInt32(0),
		enqueueLoopRunning: atomic.NewBool(false),
		syncOnCreate:       opts.syncOnCreate,
	}

	return cache, nil
}

func (w *InMemoryAutoRefresh) Start(ctx context.Context) error {
	for i := uint(0); i < w.parallelizm; i++ {