[chore] Extract queue logic into a separate internal package

bogdandrutu · bogdandrutu · commit bf090bfc8836 · 2025-06-06T15:16:22.000-07:00
Signed-off-by: Bogdan Drutu &lt;bogdandrutu@gmail.com&gt;
diff --git a/exporter/exporterhelper/internal/queuebatch/batcher.go b/exporter/exporterhelper/internal/queuebatch/batcher.go
@@ -7,14 +7,15 @@ import (
 	"context"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 	"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[T any] interface {
 	component.Component
-	Consume(context.Context, T, Done)
+	Consume(context.Context, T, queue.Done)
 }
 
 type batcherSettings[T any] struct {
diff --git a/exporter/exporterhelper/internal/queuebatch/disabled_batcher.go b/exporter/exporterhelper/internal/queuebatch/disabled_batcher.go
@@ -7,6 +7,7 @@ import (
 	"context"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
 )
 
@@ -18,7 +19,7 @@ type disabledBatcher[T any] struct {
 	consumeFunc sender.SendFunc[T]
 }
 
-func (db *disabledBatcher[T]) Consume(ctx context.Context, req T, done Done) {
+func (db *disabledBatcher[T]) Consume(ctx context.Context, req T, done queue.Done) {
 	done.OnDone(db.consumeFunc(ctx, req))
 }
 
diff --git a/exporter/exporterhelper/internal/queuebatch/disabled_batcher_test.go b/exporter/exporterhelper/internal/queuebatch/disabled_batcher_test.go
@@ -13,6 +13,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
 )
@@ -36,12 +37,12 @@ func TestDisabledBatcher(t *testing.T) {
 			sink := requesttest.NewSink()
 			ba := newDisabledBatcher(sink.Export)
 
-			mq := newMemoryQueue[request.Request](memoryQueueSettings[request.Request]{
-				sizer:           request.RequestsSizer[request.Request]{},
-				capacity:        1000,
-				blockOnOverflow: true,
-			})
-			q := newAsyncQueue(mq, tt.maxWorkers, ba.Consume)
+			q := queue.NewQueue[request.Request](queue.Settings[request.Request]{
+				Sizer:           request.RequestsSizer[request.Request]{},
+				Capacity:        1000,
+				BlockOnOverflow: true,
+				NumConsumers:    tt.maxWorkers,
+			}, ba.Consume)
 
 			require.NoError(t, q.Start(context.Background(), componenttest.NewNopHost()))
 			require.NoError(t, ba.Start(context.Background(), componenttest.NewNopHost()))
diff --git a/exporter/exporterhelper/internal/queuebatch/internal/queue/async_queue.go b/exporter/exporterhelper/internal/queuebatch/internal/queue/async_queue.go
@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
+package queue // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 
 import (
 	"context"
diff --git a/exporter/exporterhelper/internal/queuebatch/internal/queue/async_queue_test.go b/exporter/exporterhelper/internal/queuebatch/internal/queue/async_queue_test.go
@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package queuebatch
+package queue
 
 import (
 	"context"
diff --git a/exporter/exporterhelper/internal/queuebatch/internal/queue/cond.go b/exporter/exporterhelper/internal/queuebatch/internal/queue/cond.go
@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
+package queue // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 
 import (
 	"context"
diff --git a/exporter/exporterhelper/internal/queuebatch/internal/queue/memory_queue.go b/exporter/exporterhelper/internal/queuebatch/internal/queue/memory_queue.go
@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
+package queue // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 
 import (
 	"context"
diff --git a/exporter/exporterhelper/internal/queuebatch/internal/queue/memory_queue_test.go b/exporter/exporterhelper/internal/queuebatch/internal/queue/memory_queue_test.go
@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package queuebatch
+package queue
 
 import (
 	"context"
diff --git a/exporter/exporterhelper/internal/queuebatch/internal/queue/persistent_queue.go b/exporter/exporterhelper/internal/queuebatch/internal/queue/persistent_queue.go
@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
+package queue // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 
 import (
 	"context"
diff --git a/exporter/exporterhelper/internal/queuebatch/internal/queue/persistent_queue_test.go b/exporter/exporterhelper/internal/queuebatch/internal/queue/persistent_queue_test.go
@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package queuebatch
+package queue
 
 import (
 	"context"
diff --git a/exporter/exporterhelper/internal/queuebatch/internal/queue/queue.go b/exporter/exporterhelper/internal/queuebatch/internal/queue/queue.go
@@ -1,13 +1,15 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
+package queue // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 
 import (
 	"context"
 	"errors"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
+	"go.opentelemetry.io/collector/pipeline"
 )
 
 type Encoding[T any] interface {
@@ -51,6 +53,44 @@ type Queue[T any] interface {
 	Capacity() int64
 }
 
+// Settings define internal parameters for a new Queue creation.
+type Settings[T any] struct {
+	Sizer           request.Sizer[T]
+	SizerType       request.SizerType
+	Capacity        int64
+	NumConsumers    int
+	WaitForResult   bool
+	BlockOnOverflow bool
+	Signal          pipeline.Signal
+	StorageID       *component.ID
+	Encoding        Encoding[T]
+	ID              component.ID
+	Telemetry       component.TelemetrySettings
+}
+
+func NewQueue[T any](set Settings[T], next ConsumeFunc[T]) Queue[T] {
+	// Configure memory queue or persistent based on the config.
+	if set.StorageID == nil {
+		return newAsyncQueue(newMemoryQueue[T](memoryQueueSettings[T]{
+			sizer:           set.Sizer,
+			capacity:        set.Capacity,
+			waitForResult:   set.WaitForResult,
+			blockOnOverflow: set.BlockOnOverflow,
+		}), set.NumConsumers, next)
+	}
+	return newAsyncQueue(newPersistentQueue[T](persistentQueueSettings[T]{
+		sizer:           set.Sizer,
+		sizerType:       set.SizerType,
+		capacity:        set.Capacity,
+		blockOnOverflow: set.BlockOnOverflow,
+		signal:          set.Signal,
+		storageID:       *set.StorageID,
+		encoding:        set.Encoding,
+		id:              set.ID,
+		telemetry:       set.Telemetry,
+	}), set.NumConsumers, next)
+}
+
 // TODO: Investigate why linter "unused" fails if add a private "read" func on the Queue.
 type readableQueue[T any] interface {
 	Queue[T]
diff --git a/exporter/exporterhelper/internal/queuebatch/multi_batcher.go b/exporter/exporterhelper/internal/queuebatch/multi_batcher.go
@@ -7,6 +7,7 @@ import (
 	"sync"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
 )
@@ -62,7 +63,7 @@ func (mb *multiBatcher) Start(context.Context, component.Host) error {
 	return nil
 }
 
-func (mb *multiBatcher) Consume(ctx context.Context, req request.Request, done Done) {
+func (mb *multiBatcher) Consume(ctx context.Context, req request.Request, done queue.Done) {
 	shard := mb.getPartition(ctx, req)
 	shard.Consume(ctx, req, done)
 }
diff --git a/exporter/exporterhelper/internal/queuebatch/obs_queue.go b/exporter/exporterhelper/internal/queuebatch/obs_queue.go
@@ -11,6 +11,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadata"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/pipeline"
 )
@@ -25,14 +26,14 @@ const (
 
 // obsQueue is a helper to add observability to a queue.
 type obsQueue[T request.Request] struct {
-	Queue[T]
+	queue.Queue[T]
 	tb                *metadata.TelemetryBuilder
 	metricAttr        metric.MeasurementOption
 	enqueueFailedInst metric.Int64Counter
 	tracer            trace.Tracer
 }
 
-func newObsQueue[T request.Request](set Settings[T], delegate Queue[T]) (Queue[T], error) {
+func newObsQueue[T request.Request](set Settings[T], delegate queue.Queue[T]) (queue.Queue[T], error) {
 	tb, err := metadata.NewTelemetryBuilder(set.Telemetry)
 	if err != nil {
 		return nil, err
diff --git a/exporter/exporterhelper/internal/queuebatch/obs_queue_test.go b/exporter/exporterhelper/internal/queuebatch/obs_queue_test.go
@@ -16,6 +16,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadatatest"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
 	"go.opentelemetry.io/collector/exporter/exportertest"
@@ -25,7 +26,7 @@ import (
 var exporterID = component.NewID(exportertest.NopType)
 
 type fakeQueue[T any] struct {
-	Queue[T]
+	queue.Queue[T]
 	offerErr error
 	size     int64
 	capacity int64
@@ -43,7 +44,7 @@ func (fq *fakeQueue[T]) Offer(context.Context, T) error {
 	return fq.offerErr
 }
 
-func newFakeQueue[T request.Request](offerErr error, size, capacity int64) Queue[T] {
+func newFakeQueue[T request.Request](offerErr error, size, capacity int64) queue.Queue[T] {
 	return &fakeQueue[T]{offerErr: offerErr, size: size, capacity: capacity}
 }
 
diff --git a/exporter/exporterhelper/internal/queuebatch/partition_batcher.go b/exporter/exporterhelper/internal/queuebatch/partition_batcher.go
@@ -11,6 +11,7 @@ import (
 	"go.uber.org/multierr"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
 )
@@ -60,7 +61,7 @@ func (qb *partitionBatcher) resetTimer() {
 	}
 }
 
-func (qb *partitionBatcher) Consume(ctx context.Context, req request.Request, done Done) {
+func (qb *partitionBatcher) Consume(ctx context.Context, req request.Request, done queue.Done) {
 	qb.currentBatchMu.Lock()
 
 	if qb.currentBatch == nil {
@@ -203,7 +204,7 @@ func (qb *partitionBatcher) flushCurrentBatchIfNecessary() {
 }
 
 // flush starts a goroutine that calls consumeFunc. It blocks until a worker is available if necessary.
-func (qb *partitionBatcher) flush(ctx context.Context, req request.Request, done Done) {
+func (qb *partitionBatcher) flush(ctx context.Context, req request.Request, done queue.Done) {
 	qb.stopWG.Add(1)
 	qb.wp.execute(func() {
 		defer qb.stopWG.Done()
@@ -229,7 +230,7 @@ func (wp *workerPool) execute(f func()) {
 	wp.workers <- struct{}{}
 }
 
-type multiDone []Done
+type multiDone []queue.Done
 
 func (mdc multiDone) OnDone(err error) {
 	for _, d := range mdc {
@@ -238,13 +239,13 @@ func (mdc multiDone) OnDone(err error) {
 }
 
 type refCountDone struct {
-	done     Done
+	done     queue.Done
 	mu       sync.Mutex
 	refCount int64
 	err      error
 }
 
-func newRefCountDone(done Done, refCount int64) Done {
+func newRefCountDone(done queue.Done, refCount int64) queue.Done {
 	return &refCountDone{
 		done:     done,
 		refCount: refCount,
diff --git a/exporter/exporterhelper/internal/queuebatch/queue_batch.go b/exporter/exporterhelper/internal/queuebatch/queue_batch.go
@@ -9,6 +9,7 @@ import (
 	"fmt"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
 	"go.opentelemetry.io/collector/pipeline"
@@ -19,13 +20,13 @@ type Settings[T any] struct {
 	Signal      pipeline.Signal
 	ID          component.ID
 	Telemetry   component.TelemetrySettings
-	Encoding    Encoding[T]
+	Encoding    queue.Encoding[T]
 	Sizers      map[request.SizerType]request.Sizer[T]
 	Partitioner Partitioner[T]
 }
 
 type QueueBatch struct {
-	queue   Queue[request.Request]
+	queue   queue.Queue[request.Request]
 	batcher Batcher[request.Request]
 }
 
@@ -81,28 +82,19 @@ func newQueueBatch(
 		cfg.NumConsumers = 1
 	}
 
-	var q Queue[request.Request]
-	// Configure memory queue or persistent based on the config.
-	if cfg.StorageID == nil {
-		q = newAsyncQueue(newMemoryQueue[request.Request](memoryQueueSettings[request.Request]{
-			sizer:           sizer,
-			capacity:        cfg.QueueSize,
-			waitForResult:   cfg.WaitForResult,
-			blockOnOverflow: cfg.BlockOnOverflow,
-		}), cfg.NumConsumers, b.Consume)
-	} else {
-		q = newAsyncQueue(newPersistentQueue[request.Request](persistentQueueSettings[request.Request]{
-			sizer:           sizer,
-			sizerType:       cfg.Sizer,
-			capacity:        cfg.QueueSize,
-			blockOnOverflow: cfg.BlockOnOverflow,
-			signal:          set.Signal,
-			storageID:       *cfg.StorageID,
-			encoding:        set.Encoding,
-			id:              set.ID,
-			telemetry:       set.Telemetry,
-		}), cfg.NumConsumers, b.Consume)
-	}
+	q := queue.NewQueue[request.Request](queue.Settings[request.Request]{
+		Sizer:           sizer,
+		SizerType:       cfg.Sizer,
+		Capacity:        cfg.QueueSize,
+		NumConsumers:    cfg.NumConsumers,
+		WaitForResult:   cfg.WaitForResult,
+		BlockOnOverflow: cfg.BlockOnOverflow,
+		Signal:          set.Signal,
+		StorageID:       cfg.StorageID,
+		Encoding:        set.Encoding,
+		ID:              set.ID,
+		Telemetry:       set.Telemetry,
+	}, b.Consume)
 
 	oq, err := newObsQueue(set, q)
 	if err != nil {
diff --git a/exporter/exporterhelper/internal/queuebatch/queue_batch_test.go b/exporter/exporterhelper/internal/queuebatch/queue_batch_test.go
@@ -18,6 +18,7 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/experr"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/hosttest"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sendertest"
@@ -52,7 +53,7 @@ func (f fakeEncoding) Unmarshal([]byte) (request.Request, error) {
 	return f.mr, nil
 }
 
-func newFakeEncoding(mr request.Request) Encoding[request.Request] {
+func newFakeEncoding(mr request.Request) queue.Encoding[request.Request] {
 	return &fakeEncoding{mr: mr}
 }
 

Original file line number	Diff line number	Diff line change
`@@ -7,14 +7,15 @@ import (`
`7`	`7`	`"context"`
`8`	`8`
`9`	`9`	`"go.opentelemetry.io/collector/component"`
	`10`	`+ "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"`
`10`	`11`	`"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"`
`11`	`12`	`"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"`
`12`	`13`	`)`
`13`	`14`
`14`	`15`	`// Batcher is in charge of reading items from the queue and send them out asynchronously.`
`15`	`16`	`type Batcher[T any] interface {`
`16`	`17`	`component.Component`
`17`		`- Consume(context.Context, T, Done)`
	`18`	`+ Consume(context.Context, T, queue.Done)`
`18`	`19`	`}`
`19`	`20`
`20`	`21`	`type batcherSettings[T any] struct {`
Original file line number	Diff line number	Diff line change
`@@ -7,6 +7,7 @@ import (`
`7`	`7`	`"context"`
`8`	`8`
`9`	`9`	`"go.opentelemetry.io/collector/component"`
	`10`	`+ "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch/internal/queue"`
`10`	`11`	`"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"`
`11`	`12`	`)`
`12`	`13`
`@@ -18,7 +19,7 @@ type disabledBatcher[T any] struct {`
`18`	`19`	`consumeFunc sender.SendFunc[T]`
`19`	`20`	`}`
`20`	`21`
`21`		`-func (db *disabledBatcher[T]) Consume(ctx context.Context, req T, done Done) {`
	`22`	`+func (db *disabledBatcher[T]) Consume(ctx context.Context, req T, done queue.Done) {`
`22`	`23`	`done.OnDone(db.consumeFunc(ctx, req))`
`23`	`24`	`}`
`24`	`25`