chore: refactor background flushing

Makefile

@@ -3,7 +3,7 @@
 all: test lint
 
 test:
-	go test -timeout 10m -v -race ./...
+	go test -timeout 1m -v -race ./...
 	go vet ./...
 
 godoc:

internal/background_flusher.go

@@ -0,0 +1,55 @@
+package internal
+
+import (
+	"log"
+	"time"
+)
+
+type BackgroundFlusher interface {
+	Start()
+	Stop()
+}
+
+type backgroundFlusher struct {
+	ticker   *time.Ticker
+	interval time.Duration
+	handler  LineHandler
+	stop     chan struct{}
+}
+
+func NewBackgroundFlusher(interval time.Duration, handler LineHandler) BackgroundFlusher {
+	return &backgroundFlusher{
+		interval: interval,
+		handler:  handler,
+		stop:     make(chan struct{}),
+	}
+}
+
+func (f *backgroundFlusher) Start() {
+	format := f.handler.Format()
+	if f.ticker != nil {
+		return
+	}
+	f.ticker = time.NewTicker(f.interval)
+	go func() {
+		for {
+			select {
+			case tick := <-f.ticker.C:
+				log.Printf("%s -- flushing at: %s\n", format, tick)
+				err := f.handler.FlushWithThrottling()
+				if err != nil {
+					log.Printf("%s -- error during background flush: %s\n", format, err.Error())
+				} else {
+					log.Printf("%s -- flush completed at %s\n", format, time.Now())
+				}
+			case <-f.stop:
+				return
+			}
+		}
+	}()
+}
+
+func (f *backgroundFlusher) Stop() {
+	f.ticker.Stop()
+	f.stop <- struct{}{}
+}

internal/handler_factory.go

@@ -80,7 +80,7 @@ func (f *HandlerFactory) NewSpanLogHandler(batchSize int) *RealLineHandler {
 }
 
 // NewEventHandler creates a RealLineHandler for the Event type
-// The Event handler always sets "SetLockOnThrottledError" to true
+// The Event handler always sets "ThrottleRequestsOnBackpressure" to true
 // And always uses a batch size of exactly 1.
 func (f *HandlerFactory) NewEventHandler() *RealLineHandler {
 	return NewLineHandler(
@@ -91,6 +91,6 @@ func (f *HandlerFactory) NewEventHandler() *RealLineHandler {
 		f.bufferSize,
 		append(f.lineHandlerOptions,
 			SetHandlerPrefix("events"),
-			SetLockOnThrottledError(true))...,
+			ThrottleRequestsOnBackpressure())...,
 	)
 }

internal/interfaces.go

@@ -29,7 +29,9 @@ type LineHandler interface {
 	Start()
 	Stop()
 	Flush() error
+	FlushWithThrottling() error
 	GetFailureCount() int64
+	Format() string
 }
 
 const (