Define GroupAggregator interface

2025-10-26 11:16:12 +08:00 · 2022-04-07 06:13:49 -07:00
parent 7a10e61030
commit 2471f1ebe2
3 changed files with 55 additions and 39 deletions
--- a/aggregator_test.go
+++ b/aggregator_test.go
@@ -120,13 +120,13 @@ func TestAggregator(t *testing.T) {
 		h.FlushDB(t, r)

 		aggregator := newAggregator(aggregatorParams{
-			logger:        testLogger,
-			broker:        rdbClient,
-			queues:        []string{"default"},
-			gracePeriod:   tc.gracePeriod,
-			maxDelay:      tc.maxDelay,
-			maxSize:       tc.maxSize,
-			aggregateFunc: tc.aggregateFunc,
+			logger:          testLogger,
+			broker:          rdbClient,
+			queues:          []string{"default"},
+			gracePeriod:     tc.gracePeriod,
+			maxDelay:        tc.maxDelay,
+			maxSize:         tc.maxSize,
+			groupAggregator: GroupAggregatorFunc(tc.aggregateFunc),
 		})

 		var wg sync.WaitGroup