Configure cleanQueueInterval

Signed-off-by: Thomas Poignant <[email protected]>

Author: thomaspoignant

cmd/relayproxy/config/config.go

@@ -259,6 +259,11 @@ type Config struct {
 	// Exporters is the exact same things than Exporter but allows to give more than 1 exporter at the time.
 	Exporters *[]ExporterConf `mapstructure:"exporters" koanf:"exporters"`
 
+	// ExporterCleanQueueInterval (optional) is the duration between each cleaning of the queue by the thread in charge
+	// of removing the old events.
+	// Default: 1 minute
+	ExporterCleanQueueInterval time.Duration `mapstructure:"exporterCleanQueueInterval" koanf:"exportercleanqueueinterval"`
+
 	// Notifiers is the configuration on where to notify a flag change
 	Notifiers []NotifierConf `mapstructure:"notifier" koanf:"notifier"`
 

config.go

@@ -78,6 +78,11 @@ type Config struct {
 	// Multiple exporters can be used to send data to multiple destinations in parallel without interference.
 	DataExporters []DataExporter
 
+	// ExporterCleanQueueInterval (optional) is the duration between each cleaning of the queue by the thread in charge
+	// of removing the old events.
+	// Default: 1 minute
+	ExporterCleanQueueInterval time.Duration
+
 	// StartWithRetrieverError (optional) If true, the SDK will start even if we did not get any flags from the retriever.
 	// It will serve only default values until all the retrievers returns the flags.
 	// The init method will not return any error if the flag file is unreachable.

exporter/event_store.go

@@ -20,17 +20,17 @@ type eventStoreImpl[T any] struct {
 	lastOffset int64
 	// stopPeriodicCleaning is a channel to stop the periodic cleaning goroutine
 	stopPeriodicCleaning chan struct{}
-	// cleanQueueDuration is the duration between each cleaning
-	cleanQueueDuration time.Duration
+	// cleanQueueInterval is the duration between each cleaning
+	cleanQueueInterval time.Duration
 }
 
-func NewEventStore[T any](cleanQueueDuration time.Duration) EventStore[T] {
+func NewEventStore[T any](cleanQueueInterval time.Duration) EventStore[T] {
 	store := &eventStoreImpl[T]{
 		events:               make([]Event[T], 0),
 		mutex:                sync.RWMutex{},
 		lastOffset:           minOffset,
 		stopPeriodicCleaning: make(chan struct{}),
-		cleanQueueDuration:   cleanQueueDuration,
+		cleanQueueInterval:   cleanQueueInterval,
 		consumers:            make(map[string]*consumer),
 	}
 	go store.periodicCleanQueue()
@@ -166,7 +166,7 @@ func (e *eventStoreImpl[T]) cleanQueue() {
 
 // periodicCleanQueue periodically cleans the queue
 func (e *eventStoreImpl[T]) periodicCleanQueue() {
-	ticker := time.NewTicker(e.cleanQueueDuration)
+	ticker := time.NewTicker(e.cleanQueueInterval)
 	defer ticker.Stop()
 	for {
 		select {

exporter/manager.go

@@ -20,12 +20,17 @@ type managerImpl[T any] struct {
 	eventStore *EventStore[T]
 }
 
-func NewManager[T any](ctx context.Context, exporters []Config, logger *fflog.FFLogger) Manager[T] {
+func NewManager[T any](ctx context.Context, exporters []Config, ExporterCleanQueueInterval time.Duration, logger *fflog.FFLogger) Manager[T] {
 	if ctx == nil {
 		ctx = context.Background()
 	}
 
-	evStore := NewEventStore[T](30 * time.Second)
+	if ExporterCleanQueueInterval == 0 {
+		// default value for the exporterCleanQueueDuration is 1 minute
+		ExporterCleanQueueInterval = 1 * time.Minute
+	}
+
+	evStore := NewEventStore[T](ExporterCleanQueueInterval)
 	consumers := make([]DataExporter[T], len(exporters))
 	for index, exporter := range exporters {
 		consumerID := uuid.New().String()

feature_flag.go

@@ -133,8 +133,8 @@ func New(config Config) (*GoFeatureFlag, error) {
 					MaxEventInMemory: exp.MaxEventInMemory,
 				}
 			}
-			goFF.dataExporter =
-				exporter.NewManager[exporter.FeatureEvent](config.Context, expConfigs, goFF.config.internalLogger)
+			goFF.dataExporter = exporter.NewManager[exporter.FeatureEvent](
+				config.Context, expConfigs, config.ExporterCleanQueueInterval, goFF.config.internalLogger)
 			go goFF.dataExporter.Start()
 		}
 	}