Add TelemetryCounter.Stop()

counter/counter.go

@@ -21,25 +21,38 @@ import "encoding/json"
 type TelemetryCounter struct {
 	resources map[string]*ResourceCounter
 	commands  chan func()
+	done      chan struct{}
 }
 
 // NewTelemetryCounter creates a new TelemetryCounter.
 func NewTelemetryCounter() *TelemetryCounter {
 	t := &TelemetryCounter{
 		resources: make(map[string]*ResourceCounter),
 		commands:  make(chan func()),
+		done:      make(chan struct{}),
 	}
 	go t.run()
 	return t
 }
 
 // run listens for commands to modify or read the resources.
 func (t *TelemetryCounter) run() {
-	for cmd := range t.commands {
-		cmd()
+	for {
+		select {
+		case cmd := <-t.commands:
+			cmd() // Execute the command
+		case <-t.done:
+			// Shutdown signal received, exit the loop
+			return
+		}
 	}
 }
 
+// Stop gracefully shuts down the TelemetryCounter.
+func (t *TelemetryCounter) Stop() {
+	close(t.done)
+}
+
 // Add increments the counter with the supplied dimensions.
 func (t *TelemetryCounter) Add(resource, attributes map[string]any) {
 	t.commands <- func() {

processor/datapointcountprocessor/processor.go

@@ -104,6 +104,7 @@ func (p *metricCountProcessor) Shutdown(_ context.Context) error {
 		p.cancel()
 	}
 	p.wg.Wait()
+	p.counter.Stop()
 	return nil
 }
 

processor/logcountprocessor/processor.go

@@ -99,6 +99,7 @@ func (p *logCountProcessor) Shutdown(_ context.Context) error {
 		p.cancel()
 	}
 	p.wg.Wait()
+	p.counter.Stop()
 	return nil
 }
 

processor/spancountprocessor/processor.go

@@ -100,6 +100,7 @@ func (p *spanCountProcessor) Shutdown(_ context.Context) error {
 		p.cancel()
 	}
 	p.wg.Wait()
+	p.counter.Stop()
 	return nil
 }