| 112 | } |
| 113 | |
| 114 | func newChannelCache(ctx context.Context, dbName string, options ChannelCacheOptions, queryHandlerFactory ChannelQueryHandlerFactory, |
| 115 | activeChannels *channels.ActiveChannels, cacheStats *base.CacheStats) (*channelCacheImpl, error) { |
| 116 | |
| 117 | channelCache := &channelCacheImpl{ |
| 118 | queryHandlerFactory: queryHandlerFactory, |
| 119 | channelCaches: channels.NewRangeSafeCollection(), |
| 120 | dbName: dbName, |
| 121 | terminator: make(chan bool), |
| 122 | options: options, |
| 123 | maxChannels: options.MaxNumChannels, |
| 124 | compactHighWatermark: int(math.Round(float64(options.CompactHighWatermarkPercent) / 100 * float64(options.MaxNumChannels))), |
| 125 | compactLowWatermark: int(math.Round(float64(options.CompactLowWatermarkPercent) / 100 * float64(options.MaxNumChannels))), |
| 126 | activeChannels: activeChannels, |
| 127 | cacheStats: cacheStats, |
| 128 | } |
| 129 | bgt, err := NewBackgroundTask(ctx, "CleanAgedItems", channelCache.cleanAgedItems, options.ChannelCacheAge, channelCache.terminator) |
| 130 | if err != nil { |
| 131 | return nil, err |
| 132 | } |
| 133 | channelCache.backgroundTasks = append(channelCache.backgroundTasks, bgt) |
| 134 | base.DebugfCtx(ctx, base.KeyCache, "Initialized channel cache with maxChannels:%d, HWM: %d, LWM: %d", |
| 135 | channelCache.maxChannels, channelCache.compactHighWatermark, channelCache.compactLowWatermark) |
| 136 | return channelCache, nil |
| 137 | } |
| 138 | |
| 139 | func (c *channelCacheImpl) Clear() { |
| 140 | c.seqLock.Lock() |