[chore][fileconsumer] Move splitter factory code into internal package

pkg/stanza/fileconsumer/config.go

@@ -15,6 +15,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/emit"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/header"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/splitter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/matcher"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
@@ -86,7 +87,7 @@ func (c Config) Build(logger *zap.SugaredLogger, emit emit.Callback) (*Manager,
 	}
 
 	// Ensure that splitter is buildable
-	factory := newMultilineSplitterFactory(c.Splitter)
+	factory := splitter.NewMultilineFactory(c.Splitter)
 	if _, err := factory.Build(int(c.MaxLogSize)); err != nil {
 		return nil, err
 	}
@@ -105,15 +106,15 @@ func (c Config) BuildWithSplitFunc(logger *zap.SugaredLogger, emit emit.Callback
 	}
 
 	// Ensure that splitter is buildable
-	factory := newCustomizeSplitterFactory(c.Splitter.Flusher, splitFunc)
+	factory := splitter.NewCustomFactory(c.Splitter.Flusher, splitFunc)
 	if _, err := factory.Build(int(c.MaxLogSize)); err != nil {
 		return nil, err
 	}
 
 	return c.buildManager(logger, emit, factory)
 }
 
-func (c Config) buildManager(logger *zap.SugaredLogger, emit emit.Callback, factory splitterFactory) (*Manager, error) {
+func (c Config) buildManager(logger *zap.SugaredLogger, emit emit.Callback, factory splitter.Factory) (*Manager, error) {
 	if emit == nil {
 		return nil, fmt.Errorf("must provide emit function")
 	}

pkg/stanza/fileconsumer/internal/splitter/custom.go

@@ -0,0 +1,31 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package splitter // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/splitter"
+
+import (
+	"bufio"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
+)
+
+type customFactory struct {
+	Flusher  helper.FlusherConfig
+	Splitter bufio.SplitFunc
+}
+
+var _ Factory = (*customFactory)(nil)
+
+func NewCustomFactory(
+	flusher helper.FlusherConfig,
+	splitter bufio.SplitFunc) Factory {
+	return &customFactory{
+		Flusher:  flusher,
+		Splitter: splitter,
+	}
+}
+
+// Build builds Multiline Splitter struct
+func (factory *customFactory) Build(_ int) (bufio.SplitFunc, error) {
+	return factory.Flusher.Build().SplitFunc(factory.Splitter), nil
+}

pkg/stanza/fileconsumer/internal/splitter/custom_test.go

@@ -0,0 +1,56 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package splitter
+
+import (
+	"bufio"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
+)
+
+func TestCustomFactory(t *testing.T) {
+	type fields struct {
+		Flusher  helper.FlusherConfig
+		Splitter bufio.SplitFunc
+	}
+	type args struct {
+		maxLogSize int
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		args    args
+		wantErr bool
+	}{
+		{
+			name: "default configuration",
+			fields: fields{
+				Flusher: helper.NewFlusherConfig(),
+				Splitter: func(data []byte, atEOF bool) (advance int, token []byte, err error) {
+					return len(data), data, nil
+				},
+			},
+			args: args{
+				maxLogSize: 1024,
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			factory := NewCustomFactory(tt.fields.Flusher, tt.fields.Splitter)
+			got, err := factory.Build(tt.args.maxLogSize)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("Build() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if err == nil {
+				assert.NotNil(t, got)
+			}
+		})
+	}
+}

pkg/stanza/fileconsumer/internal/splitter/factory.go

@@ -0,0 +1,12 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package splitter // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/splitter"
+
+import (
+	"bufio"
+)
+
+type Factory interface {
+	Build(maxLogSize int) (bufio.SplitFunc, error)
+}

pkg/stanza/fileconsumer/internal/splitter/multiline.go

@@ -0,0 +1,36 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package splitter // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/splitter"
+
+import (
+	"bufio"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
+)
+
+type multilineFactory struct {
+	helper.SplitterConfig
+}
+
+var _ Factory = (*multilineFactory)(nil)
+
+func NewMultilineFactory(splitter helper.SplitterConfig) Factory {
+	return &multilineFactory{
+		SplitterConfig: splitter,
+	}
+}
+
+// Build builds Multiline Splitter struct
+func (factory *multilineFactory) Build(maxLogSize int) (bufio.SplitFunc, error) {
+	enc, err := helper.LookupEncoding(factory.EncodingConfig.Encoding)
+	if err != nil {
+		return nil, err
+	}
+	flusher := factory.Flusher.Build()
+	splitter, err := factory.Multiline.Build(enc, false, factory.PreserveLeadingWhitespaces, factory.PreserveTrailingWhitespaces, flusher, maxLogSize)
+	if err != nil {
+		return nil, err
+	}
+	return splitter, nil
+}

pkg/stanza/fileconsumer/internal/splitter/multiline_test.go

@@ -0,0 +1,75 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package splitter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
+)
+
+func TestMultilineBuild(t *testing.T) {
+	type args struct {
+		maxLogSize int
+	}
+	tests := []struct {
+		name           string
+		splitterConfig helper.SplitterConfig
+		args           args
+		wantErr        bool
+	}{
+		{
+			name:           "default configuration",
+			splitterConfig: helper.NewSplitterConfig(),
+			args: args{
+				maxLogSize: 1024,
+			},
+			wantErr: false,
+		},
+		{
+			name: "eoncoding error",
+			splitterConfig: helper.SplitterConfig{
+				EncodingConfig: helper.EncodingConfig{
+					Encoding: "error",
+				},
+				Flusher:   helper.NewFlusherConfig(),
+				Multiline: helper.NewMultilineConfig(),
+			},
+			args: args{
+				maxLogSize: 1024,
+			},
+			wantErr: true,
+		},
+		{
+			name: "Multiline  error",
+			splitterConfig: helper.SplitterConfig{
+				EncodingConfig: helper.NewEncodingConfig(),
+				Flusher:        helper.NewFlusherConfig(),
+				Multiline: helper.MultilineConfig{
+					LineStartPattern: "START",
+					LineEndPattern:   "END",
+				},
+			},
+			args: args{
+				maxLogSize: 1024,
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			factory := NewMultilineFactory(tt.splitterConfig)
+			got, err := factory.Build(tt.args.maxLogSize)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("Build() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if err == nil {
+				assert.NotNil(t, got)
+			}
+		})
+	}
+}

pkg/stanza/fileconsumer/reader_factory.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/header"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/splitter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/util"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
 )
@@ -21,7 +22,7 @@ type readerFactory struct {
 	*zap.SugaredLogger
 	readerConfig    *readerConfig
 	fromBeginning   bool
-	splitterFactory splitterFactory
+	splitterFactory splitter.Factory
 	encodingConfig  helper.EncodingConfig
 	headerConfig    *header.Config
 }

pkg/stanza/fileconsumer/reader_test.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/header"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/splitter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/parser/regex"
@@ -135,7 +136,7 @@ func TestTokenizationTooLongWithLineStartPattern(t *testing.T) {
 
 	mlc := helper.NewMultilineConfig()
 	mlc.LineStartPattern = `\d+-\d+-\d+`
-	f.splitterFactory = newMultilineSplitterFactory(helper.SplitterConfig{
+	f.splitterFactory = splitter.NewMultilineFactory(helper.SplitterConfig{
 		EncodingConfig: helper.NewEncodingConfig(),
 		Flusher:        helper.NewFlusherConfig(),
 		Multiline:      mlc,
@@ -200,7 +201,7 @@ func testReaderFactory(t *testing.T) (*readerFactory, chan *emitParams) {
 			emit:            testEmitFunc(emitChan),
 		},
 		fromBeginning:   true,
-		splitterFactory: newMultilineSplitterFactory(splitterConfig),
+		splitterFactory: splitter.NewMultilineFactory(splitterConfig),
 		encodingConfig:  splitterConfig.EncodingConfig,
 	}, emitChan
 }
@@ -229,7 +230,7 @@ func TestEncodingDecode(t *testing.T) {
 			fingerprintSize: fingerprint.DefaultSize,
 			maxLogSize:      defaultMaxLogSize,
 		},
-		splitterFactory: newMultilineSplitterFactory(helper.NewSplitterConfig()),
+		splitterFactory: splitter.NewMultilineFactory(helper.NewSplitterConfig()),
 		fromBeginning:   false,
 	}
 	r, err := f.newReader(testFile, fp)