Skip to content

Commit 2b413a2

Browse files
committed
CloudwatchService.Stop should wait for batcher
1 parent fdea1bc commit 2b413a2

File tree

2 files changed

+9
-7
lines changed

2 files changed

+9
-7
lines changed

aws/cwatch/service.go

+8-3
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,9 @@ type Service struct {
1717
Client Client
1818
namespace string
1919
deployment string
20-
batcher *syncx.Batcher[types.MetricDatum]
20+
21+
batcher *syncx.Batcher[types.MetricDatum]
22+
batcherWG *sync.WaitGroup
2123
}
2224

2325
// NewService creates a new Cloudwatch service with the given credentials and configuration. Some behaviours depend on
@@ -41,11 +43,13 @@ func NewService(accessKey, secretKey, region, namespace, deployment string) (*Se
4143
return &Service{Client: client, namespace: namespace, deployment: deployment}, nil
4244
}
4345

44-
func (s *Service) StartQueue(wg *sync.WaitGroup, maxAge time.Duration) {
46+
func (s *Service) StartQueue(maxAge time.Duration) {
4547
if s.batcher != nil {
4648
panic("queue already started")
4749
}
48-
s.batcher = syncx.NewBatcher(s.processBatch, 100, maxAge, 1000, wg)
50+
51+
s.batcherWG = &sync.WaitGroup{}
52+
s.batcher = syncx.NewBatcher(s.processBatch, 100, maxAge, 1000, s.batcherWG)
4953
s.batcher.Start()
5054
}
5155

@@ -54,6 +58,7 @@ func (s *Service) StopQueue() {
5458
panic("queue wasn't started")
5559
}
5660
s.batcher.Stop()
61+
s.batcherWG.Wait()
5762
}
5863

5964
func (s *Service) Queue(data ...types.MetricDatum) {

aws/cwatch/service_test.go

+1-4
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@ package cwatch_test
22

33
import (
44
"context"
5-
"sync"
65
"testing"
76
"time"
87

@@ -29,8 +28,7 @@ func TestService(t *testing.T) {
2928
svc, err = cwatch.NewService("root", "key", "us-east-1", "Foo", "dev")
3029
assert.NoError(t, err)
3130

32-
wg := &sync.WaitGroup{}
33-
svc.StartQueue(wg, time.Millisecond*100)
31+
svc.StartQueue(time.Millisecond * 100)
3432

3533
svc.Queue(cwatch.Datum("NumGoats", 10, types.StandardUnitCount, cwatch.Dimension("Host", "foo1")))
3634
svc.Queue(cwatch.Datum("NumSheep", 20, types.StandardUnitCount))
@@ -43,7 +41,6 @@ func TestService(t *testing.T) {
4341
svc.Queue(cwatch.Datum("SleepTime", 30, types.StandardUnitSeconds))
4442

4543
svc.StopQueue()
46-
wg.Wait()
4744

4845
// check the queued metric was sent
4946
assert.Equal(t, 2, svc.Client.(*cwatch.DevClient).CallCount())

0 commit comments

Comments
 (0)