[chore] Move obs_queue to the internal/queue package (#13363)

Updates
#13077

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

Author: bogdandrutu

exporter/exporterhelper/internal/queuebatch/obs_queue.go renamed to exporter/exporterhelper/internal/queue/obs_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/queue"
 
 import (
 	"context"
@@ -11,7 +11,6 @@ import (
 	"go.opentelemetry.io/otel/trace"
 
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadata"
-	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queue"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/pipeline"
 )
@@ -26,14 +25,14 @@ const (
 
 // obsQueue is a helper to add observability to a queue.
 type obsQueue[T request.Request] struct {
-	queue.Queue[T]
+	Queue[T]
 	tb                *metadata.TelemetryBuilder
 	metricAttr        metric.MeasurementOption
 	enqueueFailedInst metric.Int64Counter
 	tracer            trace.Tracer
 }
 
-func newObsQueue[T request.Request](set Settings[T], delegate queue.Queue[T]) (queue.Queue[T], error) {
+func newObsQueue[T request.Request](set Settings[T], delegate Queue[T]) (Queue[T], error) {
 	tb, err := metadata.NewTelemetryBuilder(set.Telemetry)
 	if err != nil {
 		return nil, err

exporter/exporterhelper/internal/queuebatch/obs_queue_test.go renamed to exporter/exporterhelper/internal/queue/obs_queue_test.go

@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package queuebatch
+package queue
 
 import (
 	"context"
@@ -16,7 +16,6 @@ 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/queue"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
 	"go.opentelemetry.io/collector/exporter/exportertest"
@@ -26,7 +25,7 @@ import (
 var exporterID = component.NewID(exportertest.NopType)
 
 type fakeQueue[T any] struct {
-	queue.Queue[T]
+	Queue[T]
 	offerErr error
 	size     int64
 	capacity int64
@@ -44,7 +43,7 @@ func (fq *fakeQueue[T]) Offer(context.Context, T) error {
 	return fq.offerErr
 }
 
-func newFakeQueue[T request.Request](offerErr error, size, capacity int64) queue.Queue[T] {
+func newFakeQueue[T request.Request](offerErr error, size, capacity int64) Queue[T] {
 	return &fakeQueue[T]{offerErr: offerErr, size: size, capacity: capacity}
 }
 

exporter/exporterhelper/internal/queue/queue.go

@@ -81,18 +81,33 @@ func (set *Settings[T]) activeSizer() request.Sizer[T] {
 	}
 }
 
-func NewQueue[T any](set Settings[T], next ConsumeFunc[T]) (Queue[T], error) {
+func NewQueue[T request.Request](set Settings[T], next ConsumeFunc[T]) (Queue[T], error) {
+	q, err := newBaseQueue(set)
+	if err != nil {
+		return nil, err
+	}
+
+	oq, err := newObsQueue(set, newAsyncQueue(q, set.NumConsumers, next))
+	if err != nil {
+		return nil, err
+	}
+
+	return oq, nil
+}
+
+func newBaseQueue[T any](set Settings[T]) (readableQueue[T], error) {
 	// Configure memory queue or persistent based on the config.
 	if !set.StorageID.HasValue() {
-		return newAsyncQueue(newMemoryQueue[T](set), set.NumConsumers, next), nil
+		return newMemoryQueue[T](set), nil
 	}
 	if set.ItemsSizer == nil {
 		return nil, errors.New("PersistentQueue requires ItemsSizer to be set")
 	}
 	if set.BytesSizer == nil {
 		return nil, errors.New("PersistentQueue requires BytesSizer to be set")
 	}
-	return newAsyncQueue(newPersistentQueue[T](set), set.NumConsumers, next), nil
+
+	return newPersistentQueue[T](set), nil
 }
 
 // TODO: Investigate why linter "unused" fails if add a private "read" func on the Queue.

exporter/exporterhelper/internal/queuebatch/disabled_batcher_test.go

@@ -43,6 +43,7 @@ func TestDisabledBatcher(t *testing.T) {
 				Capacity:        1000,
 				BlockOnOverflow: true,
 				NumConsumers:    tt.maxWorkers,
+				Telemetry:       componenttest.NewNopTelemetrySettings(),
 			}, ba.Consume)
 			require.NoError(t, err)
 

exporter/exporterhelper/internal/queuebatch/queue_batch.go

@@ -97,12 +97,7 @@ func newQueueBatch(
 		return nil, err
 	}
 
-	oq, err := newObsQueue(set, q)
-	if err != nil {
-		return nil, err
-	}
-
-	return &QueueBatch{queue: oq, batcher: b}, nil
+	return &QueueBatch{queue: q, batcher: b}, nil
 }
 
 // Start is invoked during service startup.