Try to remove pointers of state in metadata

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

Author: bogdandrutu

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

@@ -59,20 +59,16 @@ func (f *Factory) NewReader(file *os.File, fp *fingerprint.Fingerprint) (*Reader
 	m := &Metadata{
 		Fingerprint:    fp,
 		FileAttributes: attributes,
-		TokenLenState:  &tokenlen.State{},
-	}
-	if f.FlushTimeout > 0 {
-		m.FlushState = &flush.State{LastDataChange: time.Now()}
+		TokenLenState:  tokenlen.State{},
+		FlushState: flush.State{
+			FlushPeriod:    f.FlushTimeout,
+			LastDataChange: time.Now(),
+		},
 	}
 	return f.NewReaderFromMetadata(file, m)
 }
 
 func (f *Factory) NewReaderFromMetadata(file *os.File, m *Metadata) (r *Reader, err error) {
-	// Ensure TokenLenState is initialized
-	if m.TokenLenState == nil {
-		m.TokenLenState = &tokenlen.State{}
-	}
-
 	r = &Reader{
 		Metadata:             m,
 		set:                  f.TelemetrySettings,
@@ -111,7 +107,7 @@ func (f *Factory) NewReaderFromMetadata(file *os.File, m *Metadata) (r *Reader,
 	}
 
 	tokenLenFunc := m.TokenLenState.Func(f.SplitFunc)
-	flushFunc := m.FlushState.Func(tokenLenFunc, f.FlushTimeout)
+	flushFunc := m.FlushState.Func(tokenLenFunc)
 	r.contentSplitFunc = trim.WithFunc(trim.ToLength(flushFunc, f.MaxLogSize), f.TrimFunc)
 
 	if f.HeaderConfig != nil && !m.HeaderFinalized {

pkg/stanza/fileconsumer/internal/reader/reader.go

@@ -31,8 +31,8 @@ type Metadata struct {
 	RecordNum       int64
 	FileAttributes  map[string]any
 	HeaderFinalized bool
-	FlushState      *flush.State
-	TokenLenState   *tokenlen.State
+	FlushState      flush.State
+	TokenLenState   tokenlen.State
 }
 
 // Reader manages a single file

pkg/stanza/flush/flush.go

@@ -11,15 +11,16 @@ import (
 )
 
 type State struct {
+	FlushPeriod    time.Duration
 	LastDataChange time.Time
 	LastDataLength int
 }
 
 // Func wraps a bufio.SplitFunc with a timer.
 // When the timer expires, an incomplete token may be returned.
 // The timer will reset any time the data parameter changes.
-func (s *State) Func(splitFunc bufio.SplitFunc, period time.Duration) bufio.SplitFunc {
-	if s == nil || period <= 0 {
+func (s *State) Func(splitFunc bufio.SplitFunc) bufio.SplitFunc {
+	if s == nil || s.FlushPeriod <= 0 {
 		return splitFunc
 	}
 
@@ -51,7 +52,7 @@ func (s *State) Func(splitFunc bufio.SplitFunc, period time.Duration) bufio.Spli
 		}
 
 		// Flush timed out
-		if internaltime.Since(s.LastDataChange) > period {
+		if internaltime.Since(s.LastDataChange) > s.FlushPeriod {
 			s.LastDataChange = internaltime.Now()
 			s.LastDataLength = 0
 			return len(data), data, nil
@@ -61,9 +62,3 @@ func (s *State) Func(splitFunc bufio.SplitFunc, period time.Duration) bufio.Spli
 		return 0, nil, nil
 	}
 }
-
-// Deprecated: [v0.88.0] Use WithFunc instead.
-func WithPeriod(splitFunc bufio.SplitFunc, period time.Duration) bufio.SplitFunc {
-	s := &State{LastDataChange: internaltime.Now()}
-	return s.Func(splitFunc, period)
-}

pkg/stanza/flush/flush_test.go

@@ -41,9 +41,7 @@ func TestNewlineSplitFunc(t *testing.T) {
 	}
 
 	for _, tc := range testCases {
-		t.Run(tc.name+"/WithPeriod", splittest.New(WithPeriod(tc.baseFunc, tc.flushPeriod), tc.input, tc.steps...))
-
-		previousState := &State{LastDataChange: time.Now()}
-		t.Run(tc.name+"/Func", splittest.New(previousState.Func(tc.baseFunc, tc.flushPeriod), tc.input, tc.steps...))
+		previousState := &State{FlushPeriod: tc.flushPeriod, LastDataChange: time.Now()}
+		t.Run(tc.name+"/Func", splittest.New(previousState.Func(tc.baseFunc), tc.input, tc.steps...))
 	}
 }