Refactor pipelines builder

Signed-off-by: Bogdan Drutu <[email protected]>

internal/testcomponents/example_receiver_test.go

@@ -23,7 +23,11 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 )
 
+<<<<<<< HEAD
 func TestExampleReceiver(t *testing.T) {
+=======
+func TestExampleReceiverProducer(t *testing.T) {
+>>>>>>> c162af37 (Refactor pipelines builder)
 	rcv := &ExampleReceiver{}
 	host := componenttest.NewNopHost()
 	assert.False(t, rcv.Started)

service/host.go

@@ -28,9 +28,7 @@ type serviceHost struct {
 	factories         component.Factories
 	buildInfo         component.BuildInfo
 
-	builtExporters  *builder.BuiltExporters
-	builtReceivers  builder.Receivers
-	builtPipelines  builder.BuiltPipelines
+	builtPipelines  *builder.BuiltPipelines
 	builtExtensions *extensions.BuiltExtensions
 }
 
@@ -56,9 +54,9 @@ func (host *serviceHost) GetFactory(kind component.Kind, componentType config.Ty
 }
 
 func (host *serviceHost) GetExtensions() map[config.ComponentID]component.Extension {
-	return host.builtExtensions.ToMap()
+	return host.builtExtensions.GetExtensions()
 }
 
 func (host *serviceHost) GetExporters() map[config.DataType]map[config.ComponentID]component.Exporter {
-	return host.builtExporters.ToMapByDataType()
+	return host.builtPipelines.GetExporters()
 }

service/internal/builder/capabilities.go

@@ -0,0 +1,58 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package builder
+
+import (
+	"go.opentelemetry.io/collector/consumer"
+)
+
+func wrapLogs(logs consumer.Logs, cap consumer.Capabilities) consumer.Logs {
+	return capLogs{Logs: logs, cap: cap}
+}
+
+type capLogs struct {
+	consumer.Logs
+	cap consumer.Capabilities
+}
+
+func (mts capLogs) Capabilities() consumer.Capabilities {
+	return mts.cap
+}
+
+func wrapMetrics(metrics consumer.Metrics, cap consumer.Capabilities) consumer.Metrics {
+	return capMetrics{Metrics: metrics, cap: cap}
+}
+
+type capMetrics struct {
+	consumer.Metrics
+	cap consumer.Capabilities
+}
+
+func (mts capMetrics) Capabilities() consumer.Capabilities {
+	return mts.cap
+}
+
+func wrapTraces(traces consumer.Traces, cap consumer.Capabilities) consumer.Traces {
+	return capTraces{Traces: traces, cap: cap}
+}
+
+type capTraces struct {
+	consumer.Traces
+	cap consumer.Capabilities
+}
+
+func (mts capTraces) Capabilities() consumer.Capabilities {
+	return mts.cap
+}

service/internal/builder/capabilities_test.go

@@ -0,0 +1,63 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package builder
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/internal/testdata"
+)
+
+func TestWrapLogs(t *testing.T) {
+	sink := &consumertest.LogsSink{}
+	require.Equal(t, consumer.Capabilities{MutatesData: false}, sink.Capabilities())
+
+	wrap := wrapLogs(sink, consumer.Capabilities{MutatesData: true})
+	assert.Equal(t, consumer.Capabilities{MutatesData: true}, wrap.Capabilities())
+
+	assert.NoError(t, wrap.ConsumeLogs(context.Background(), testdata.GenerateLogsOneLogRecord()))
+	assert.Len(t, sink.AllLogs(), 1)
+	assert.Equal(t, testdata.GenerateLogsOneLogRecord(), sink.AllLogs()[0])
+}
+
+func TestWrapMetrics(t *testing.T) {
+	sink := &consumertest.MetricsSink{}
+	require.Equal(t, consumer.Capabilities{MutatesData: false}, sink.Capabilities())
+
+	wrap := wrapMetrics(sink, consumer.Capabilities{MutatesData: true})
+	assert.Equal(t, consumer.Capabilities{MutatesData: true}, wrap.Capabilities())
+
+	assert.NoError(t, wrap.ConsumeMetrics(context.Background(), testdata.GenerateMetricsOneMetric()))
+	assert.Len(t, sink.AllMetrics(), 1)
+	assert.Equal(t, testdata.GenerateMetricsOneMetric(), sink.AllMetrics()[0])
+}
+
+func TestWrapTraces(t *testing.T) {
+	sink := &consumertest.TracesSink{}
+	require.Equal(t, consumer.Capabilities{MutatesData: false}, sink.Capabilities())
+
+	wrap := wrapTraces(sink, consumer.Capabilities{MutatesData: true})
+	assert.Equal(t, consumer.Capabilities{MutatesData: true}, wrap.Capabilities())
+
+	assert.NoError(t, wrap.ConsumeTraces(context.Background(), testdata.GenerateTracesOneSpan()))
+	assert.Len(t, sink.AllTraces(), 1)
+	assert.Equal(t, testdata.GenerateTracesOneSpan(), sink.AllTraces()[0])
+}