[chore] Move batcher to the new QueueBatcher package (#12693)

Depends on
#12689

Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com>

Author: bogdandrutu

exporter/exporterhelper/internal/batcher/package_test.go

@@ -11,7 +11,6 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
-	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/batcher"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
@@ -41,7 +40,7 @@ func NewQueueSender(
 
 type QueueBatch struct {
 	queue   queuebatch.Queue[request.Request]
-	batcher batcher.Batcher[request.Request]
+	batcher queuebatch.Batcher[request.Request]
 }
 
 func NewQueueBatch(
@@ -50,7 +49,7 @@ func NewQueueBatch(
 	bCfg exporterbatcher.Config,
 	next sender.SendFunc[request.Request],
 ) (*QueueBatch, error) {
-	b, err := batcher.NewBatcher(bCfg, next, qCfg.NumConsumers)
+	b, err := queuebatch.NewBatcher(bCfg, next, qCfg.NumConsumers)
 	if err != nil {
 		return nil, err
 	}

exporter/exporterhelper/internal/queue_sender.go

@@ -1,22 +1,21 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package batcher // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/batcher"
+package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 
 import (
 	"context"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
-	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
 )
 
 // Batcher is in charge of reading items from the queue and send them out asynchronously.
 type Batcher[K any] interface {
 	component.Component
-	Consume(context.Context, K, queuebatch.Done)
+	Consume(context.Context, K, Done)
 }
 
 func NewBatcher(batchCfg exporterbatcher.Config, exportFunc sender.SendFunc[request.Request], maxWorkers int) (Batcher[request.Request], error) {

exporter/exporterhelper/internal/batcher/batcher.go exporter/exporterhelper/internal/queuebatch/batcher.go

@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package batcher // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/batcher"
+package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 
 import (
 	"context"
@@ -12,7 +12,6 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
-	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
 )
@@ -62,7 +61,7 @@ func (qb *defaultBatcher) resetTimer() {
 	}
 }
 
-func (qb *defaultBatcher) Consume(ctx context.Context, req request.Request, done queuebatch.Done) {
+func (qb *defaultBatcher) Consume(ctx context.Context, req request.Request, done Done) {
 	qb.currentBatchMu.Lock()
 
 	if qb.currentBatch == nil {
@@ -200,7 +199,7 @@ func (qb *defaultBatcher) flushCurrentBatchIfNecessary() {
 }
 
 // flush starts a goroutine that calls consumeFunc. It blocks until a worker is available if necessary.
-func (qb *defaultBatcher) flush(ctx context.Context, req request.Request, done queuebatch.Done) {
+func (qb *defaultBatcher) flush(ctx context.Context, req request.Request, done Done) {
 	qb.stopWG.Add(1)
 	if qb.workerPool != nil {
 		<-qb.workerPool
@@ -223,7 +222,7 @@ func (qb *defaultBatcher) Shutdown(_ context.Context) error {
 	return nil
 }
 
-type multiDone []queuebatch.Done
+type multiDone []Done
 
 func (mdc multiDone) OnDone(err error) {
 	for _, d := range mdc {
@@ -232,13 +231,13 @@ func (mdc multiDone) OnDone(err error) {
 }
 
 type refCountDone struct {
-	done     queuebatch.Done
+	done     Done
 	mu       sync.Mutex
 	refCount int64
 	err      error
 }
 
-func newRefCountDone(done queuebatch.Done, refCount int64) queuebatch.Done {
+func newRefCountDone(done Done, refCount int64) Done {
 	return &refCountDone{
 		done:     done,
 		refCount: refCount,

exporter/exporterhelper/internal/batcher/default_batcher.go exporter/exporterhelper/internal/queuebatch/default_batcher.go

@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package batcher
+package queuebatch
 
 import (
 	"context"

exporter/exporterhelper/internal/batcher/default_batcher_test.go exporter/exporterhelper/internal/queuebatch/default_batcher_test.go

@@ -1,13 +1,12 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package batcher // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/batcher"
+package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 
 import (
 	"context"
 
 	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
 )
 
@@ -19,7 +18,7 @@ type disabledBatcher[T any] struct {
 	consumeFunc sender.SendFunc[T]
 }
 
-func (db *disabledBatcher[T]) Consume(ctx context.Context, req T, done queuebatch.Done) {
+func (db *disabledBatcher[T]) Consume(ctx context.Context, req T, done Done) {
 	done.OnDone(db.consumeFunc(ctx, req))
 }
 

exporter/exporterhelper/internal/batcher/disabled_batcher.go exporter/exporterhelper/internal/queuebatch/disabled_batcher.go

@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package batcher
+package queuebatch
 
 import (
 	"context"
@@ -14,7 +14,6 @@ import (
 
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
-	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
 	"go.opentelemetry.io/collector/exporter/exporterqueue"
@@ -45,9 +44,9 @@ func TestDisabledBatcher_Basic(t *testing.T) {
 			ba, err := NewBatcher(cfg, sink.Export, tt.maxWorkers)
 			require.NoError(t, err)
 
-			q := queuebatch.NewQueue[request.Request](
+			q := NewQueue[request.Request](
 				context.Background(),
-				queuebatch.QueueSettings[request.Request]{
+				QueueSettings[request.Request]{
 					Signal:           pipeline.SignalTraces,
 					ExporterSettings: exportertest.NewNopSettings(exportertest.NopType),
 				},