Introduce additional consumer packages split by telemetry type

This change adds another set of packages representing "consumer" split by telemetry signal type. The same approach is taken as for the pdata split. Common parts are moved to internal package and then they are used from both existing combined and split consumer packages.

consumer/consumer.go

@@ -15,53 +15,23 @@
 package consumer // import "go.opentelemetry.io/collector/consumer"
 
 import (
-	"errors"
+	"go.opentelemetry.io/collector/consumer/internal"
 )
 
 // Capabilities describes the capabilities of a Processor.
-type Capabilities struct {
-	// MutatesData is set to true if Consume* function of the
-	// processor modifies the input TraceData or MetricsData argument.
-	// Processors which modify the input data MUST set this flag to true. If the processor
-	// does not modify the data it MUST set this flag to false. If the processor creates
-	// a copy of the data before modifying then this flag can be safely set to false.
-	MutatesData bool
-}
-
-type baseConsumer interface {
-	Capabilities() Capabilities
-}
-
-var errNilFunc = errors.New("nil consumer func")
-
-type baseImpl struct {
-	capabilities Capabilities
-}
+type Capabilities = internal.Capabilities
 
 // Option to construct new consumers.
-type Option func(*baseImpl)
+type Option = internal.Option
 
 // WithCapabilities overrides the default GetCapabilities function for a processor.
 // The default GetCapabilities function returns mutable capabilities.
-func WithCapabilities(capabilities Capabilities) Option {
-	return func(o *baseImpl) {
-		o.capabilities = capabilities
-	}
-}
+var WithCapabilities = internal.WithCapabilities
 
-// Capabilities implementation of the base
-func (bs baseImpl) Capabilities() Capabilities {
-	return bs.capabilities
-}
+type baseConsumer = internal.BaseConsumer
 
-func newBaseImpl(options ...Option) *baseImpl {
-	bs := &baseImpl{
-		capabilities: Capabilities{MutatesData: false},
-	}
+var errNilFunc = internal.ErrNilFunc
 
-	for _, op := range options {
-		op(bs)
-	}
+type baseImpl = internal.BaseImpl
 
-	return bs
-}
+var newBaseImpl = internal.NewBaseImpl

consumer/internal/consumer.go

@@ -0,0 +1,67 @@
+// 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 internal // import "go.opentelemetry.io/collector/consumer/internal"
+
+import (
+	"errors"
+)
+
+// Capabilities describes the capabilities of a Processor.
+type Capabilities struct {
+	// MutatesData is set to true if Consume* function of the
+	// processor modifies the input TraceData or MetricsData argument.
+	// Processors which modify the input data MUST set this flag to true. If the processor
+	// does not modify the data it MUST set this flag to false. If the processor creates
+	// a copy of the data before modifying then this flag can be safely set to false.
+	MutatesData bool
+}
+
+type BaseConsumer interface {
+	Capabilities() Capabilities
+}
+
+var ErrNilFunc = errors.New("nil consumer func")
+
+type BaseImpl struct {
+	Cap Capabilities
+}
+
+// Option to construct new consumers.
+type Option func(*BaseImpl)
+
+// WithCapabilities overrides the default GetCapabilities function for a processor.
+// The default GetCapabilities function returns mutable capabilities.
+func WithCapabilities(capabilities Capabilities) Option {
+	return func(o *BaseImpl) {
+		o.Cap = capabilities
+	}
+}
+
+// Capabilities implementation of the base
+func (bs BaseImpl) Capabilities() Capabilities {
+	return bs.Cap
+}
+
+func NewBaseImpl(options ...Option) *BaseImpl {
+	bs := &BaseImpl{
+		Cap: Capabilities{MutatesData: false},
+	}
+
+	for _, op := range options {
+		op(bs)
+	}
+
+	return bs
+}

consumer/logs/consumer.go

@@ -0,0 +1,54 @@
+// 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 logs // import "go.opentelemetry.io/collector/consumer/logs"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/consumer/internal"
+	"go.opentelemetry.io/collector/model/pdata/logs"
+)
+
+// Consumer is an interface that receives logs.Logs, processes it
+// as needed, and sends it to the next processing node if any or to the destination.
+type Consumer interface {
+	internal.BaseConsumer
+	// Consume receives logs.Logs for consumption.
+	Consume(ctx context.Context, ld logs.Logs) error
+}
+
+// ConsumeFunc is a helper function that is similar to Consume.
+type ConsumeFunc func(ctx context.Context, ld logs.Logs) error
+
+// Consume calls f(ctx, ld).
+func (f ConsumeFunc) Consume(ctx context.Context, ld logs.Logs) error {
+	return f(ctx, ld)
+}
+
+type base struct {
+	*internal.BaseImpl
+	ConsumeFunc
+}
+
+// NewConsumer returns a Logs configured with the provided options.
+func NewConsumer(consume ConsumeFunc, options ...internal.Option) (Consumer, error) {
+	if consume == nil {
+		return nil, internal.ErrNilFunc
+	}
+	return &base{
+		BaseImpl:    internal.NewBaseImpl(options...),
+		ConsumeFunc: consume,
+	}, nil
+}

consumer/metrics/consumer.go

@@ -0,0 +1,54 @@
+// 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 metrics // import "go.opentelemetry.io/collector/consumer/metrics"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/consumer/internal"
+	"go.opentelemetry.io/collector/model/pdata/metrics"
+)
+
+// Consumer is an interface that receives metrics.Metrics, processes it
+// as needed, and sends it to the next processing node if any or to the destination.
+type Consumer interface {
+	internal.BaseConsumer
+	// Consume receives metrics.Metrics for consumption.
+	Consume(ctx context.Context, ld metrics.Metrics) error
+}
+
+// ConsumeFunc is a helper function that is similar to Consume.
+type ConsumeFunc func(ctx context.Context, ld metrics.Metrics) error
+
+// Consume calls f(ctx, ld).
+func (f ConsumeFunc) Consume(ctx context.Context, ld metrics.Metrics) error {
+	return f(ctx, ld)
+}
+
+type base struct {
+	*internal.BaseImpl
+	ConsumeFunc
+}
+
+// NewConsumer returns a Metrics configured with the provided options.
+func NewConsumer(consume ConsumeFunc, options ...internal.Option) (Consumer, error) {
+	if consume == nil {
+		return nil, internal.ErrNilFunc
+	}
+	return &base{
+		BaseImpl:    internal.NewBaseImpl(options...),
+		ConsumeFunc: consume,
+	}, nil
+}

consumer/traces/consumer.go

@@ -0,0 +1,54 @@
+// 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 traces // import "go.opentelemetry.io/collector/consumer/traces"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/consumer/internal"
+	"go.opentelemetry.io/collector/model/pdata/traces"
+)
+
+// Consumer is an interface that receives traces.Traces, processes it
+// as needed, and sends it to the next processing node if any or to the destination.
+type Consumer interface {
+	internal.BaseConsumer
+	// Consume receives traces.Traces for consumption.
+	Consume(ctx context.Context, ld traces.Traces) error
+}
+
+// ConsumeFunc is a helper function that is similar to Consume.
+type ConsumeFunc func(ctx context.Context, ld traces.Traces) error
+
+// Consume calls f(ctx, ld).
+func (f ConsumeFunc) Consume(ctx context.Context, ld traces.Traces) error {
+	return f(ctx, ld)
+}
+
+type base struct {
+	*internal.BaseImpl
+	ConsumeFunc
+}
+
+// NewConsumer returns a Traces configured with the provided options.
+func NewConsumer(consume ConsumeFunc, options ...internal.Option) (Consumer, error) {
+	if consume == nil {
+		return nil, internal.ErrNilFunc
+	}
+	return &base{
+		BaseImpl:    internal.NewBaseImpl(options...),
+		ConsumeFunc: consume,
+	}, nil
+}