don't export workers

backup_handlers.go

@@ -51,32 +51,32 @@ func newBackupHandlerBase(config *BackupBaseConfig, ac *a.Client, namespace stri
 	return handler
 }
 
-func (bh *backupHandlerBase) run(ctx context.Context, writers []*WriteWorker[*models.Token]) error {
+func (bh *backupHandlerBase) run(ctx context.Context, writers []*writeWorker[*models.Token]) error {
 	readWorkers := make([]pipeline.Worker[*models.Token], bh.config.Parallel)
 	for i := 0; i < bh.config.Parallel; i++ {
 		begin := (i * PARTITIONS) / bh.config.Parallel
 		count := PARTITIONS / bh.config.Parallel // TODO verify no off by 1 error
 
-		ARRCFG := ARRConfig{
+		ARRCFG := arrConfig{
 			Namespace:      bh.namespace,
 			Set:            bh.config.Set,
 			FirstPartition: begin,
 			NumPartitions:  count,
 		}
 
-		recordReader := NewAerospikeRecordReader(
+		recordReader := newAerospikeRecordReader(
 			bh.aerospikeClient,
 			ARRCFG,
 			bh.config.Policies.ScanPolicy,
 		)
 
-		readWorkers[i] = NewReadWorker(recordReader)
+		readWorkers[i] = newReadWorker(recordReader)
 	}
 
 	processorWorkers := make([]pipeline.Worker[*models.Token], bh.config.Parallel)
 	for i := 0; i < bh.config.Parallel; i++ {
-		processor := NewNoOpProcessor()
-		processorWorkers[i] = NewProcessorWorker(processor)
+		processor := newNoOpProcessor()
+		processorWorkers[i] = newProcessorWorker(processor)
 	}
 
 	writeWorkers := make([]pipeline.Worker[*models.Token], len(writers))
@@ -133,7 +133,7 @@ func (bwh *BackupHandler) run(ctx context.Context, writers []io.Writer) {
 		defer handlePanic(errChan)
 
 		batchSize := bwh.config.Parallel
-		dataWriters := []*WriteWorker[*models.Token]{}
+		dataWriters := []*writeWorker[*models.Token]{}
 
 		for i, writer := range writers {
 
@@ -178,27 +178,27 @@ func (bwh *BackupHandler) Wait(ctx context.Context) error {
 	}
 }
 
-func getDataWriter(eb EncoderBuilder, w io.Writer, namespace string, first bool) (*WriteWorker[*models.Token], error) {
+func getDataWriter(eb EncoderBuilder, w io.Writer, namespace string, first bool) (*writeWorker[*models.Token], error) {
 	enc, err := eb.CreateEncoder()
 	if err != nil {
 		return nil, err
 	}
 
 	switch encT := enc.(type) {
 	case *asb.Encoder:
-		asbw := NewASBWriter(encT, w)
+		asbw := newAsbWriter(encT, w)
 		err := asbw.Init(namespace, first)
 		if err != nil {
 			return nil, err
 		}
 
-		worker := NewWriteWorker(asbw)
+		worker := newWriteWorker(asbw)
 
 		return worker, err
 
 	default:
-		gw := NewGenericWriter(encT, w)
-		worker := NewWriteWorker(gw)
+		gw := newGenericWriter(encT, w)
+		worker := newWriteWorker(gw)
 
 		return worker, nil
 	}

processors.go

@@ -25,40 +25,40 @@ import (
 
 // **** Processor Worker ****
 
-// DataProcessor is an interface for processing data
+// dataProcessor is an interface for processing data
 //
 //go:generate mockery --name DataProcessor
-type DataProcessor[T any] interface {
+type dataProcessor[T any] interface {
 	Process(T) (T, error)
 }
 
-// ProcessorWorker implements the pipeline.Worker interface
+// processorWorker implements the pipeline.Worker interface
 // It wraps a DataProcessor and processes data with it
-type ProcessorWorker[T any] struct {
-	processor DataProcessor[T]
+type processorWorker[T any] struct {
+	processor dataProcessor[T]
 	receive   <-chan T
 	send      chan<- T
 }
 
-// NewProcessorWorker creates a new ProcessorWorker
-func NewProcessorWorker[T any](processor DataProcessor[T]) *ProcessorWorker[T] {
-	return &ProcessorWorker[T]{
+// newProcessorWorker creates a new ProcessorWorker
+func newProcessorWorker[T any](processor dataProcessor[T]) *processorWorker[T] {
+	return &processorWorker[T]{
 		processor: processor,
 	}
 }
 
 // SetReceiveChan sets the receive channel for the ProcessorWorker
-func (w *ProcessorWorker[T]) SetReceiveChan(c <-chan T) {
+func (w *processorWorker[T]) SetReceiveChan(c <-chan T) {
 	w.receive = c
 }
 
 // SetSendChan sets the send channel for the ProcessorWorker
-func (w *ProcessorWorker[T]) SetSendChan(c chan<- T) {
+func (w *processorWorker[T]) SetSendChan(c chan<- T) {
 	w.send = c
 }
 
 // Run starts the ProcessorWorker
-func (w *ProcessorWorker[T]) Run(ctx context.Context) error {
+func (w *processorWorker[T]) Run(ctx context.Context) error {
 	for {
 		select {
 		case <-ctx.Done():
@@ -84,16 +84,16 @@ func (w *ProcessorWorker[T]) Run(ctx context.Context) error {
 
 // **** NoOp Processor ****
 
-// NoOpProcessor satisfies the DataProcessor interface
+// noOpProcessor satisfies the DataProcessor interface
 // It does nothing to the data
-type NoOpProcessor struct{}
+type noOpProcessor struct{}
 
-// NewNoOpProcessor creates a new NOOPProcessor
-func NewNoOpProcessor() *NoOpProcessor {
-	return &NoOpProcessor{}
+// newNoOpProcessor creates a new NOOPProcessor
+func newNoOpProcessor() *noOpProcessor {
+	return &noOpProcessor{}
 }
 
 // Process processes the data
-func (p *NoOpProcessor) Process(data *models.Token) (*models.Token, error) {
+func (p *noOpProcessor) Process(data *models.Token) (*models.Token, error) {
 	return data, nil
 }

processors_test.go

@@ -34,7 +34,7 @@ func (suite *proccessorTestSuite) TestProcessorWorker() {
 	mockProcessor := mocks.NewDataProcessor[string](suite.T())
 	mockProcessor.EXPECT().Process("test").Return("test", nil)
 
-	worker := NewProcessorWorker(mockProcessor)
+	worker := newProcessorWorker(mockProcessor)
 	suite.NotNil(worker)
 
 	receiver := make(chan string, 1)
@@ -58,7 +58,7 @@ func (suite *proccessorTestSuite) TestProcessorWorkerCancelOnReceive() {
 	mockProcessor := mocks.NewDataProcessor[string](suite.T())
 	mockProcessor.EXPECT().Process("test").Return("test", nil)
 
-	worker := NewProcessorWorker(mockProcessor)
+	worker := newProcessorWorker(mockProcessor)
 	suite.NotNil(worker)
 
 	receiver := make(chan string, 1)
@@ -96,7 +96,7 @@ func (suite *proccessorTestSuite) TestProcessorWorkerCancelOnSend() {
 	mockProcessor := mocks.NewDataProcessor[string](suite.T())
 	mockProcessor.EXPECT().Process("test").Return("test", nil)
 
-	worker := NewProcessorWorker(mockProcessor)
+	worker := newProcessorWorker(mockProcessor)
 	suite.NotNil(worker)
 
 	receiver := make(chan string, 1)
@@ -131,7 +131,7 @@ func (suite *proccessorTestSuite) TestProcessorWorkerReceiveClosed() {
 	mockProcessor := mocks.NewDataProcessor[string](suite.T())
 	mockProcessor.EXPECT().Process("test").Return("test", nil)
 
-	worker := NewProcessorWorker(mockProcessor)
+	worker := newProcessorWorker(mockProcessor)
 	suite.NotNil(worker)
 
 	receiver := make(chan string, 1)
@@ -152,7 +152,7 @@ func (suite *proccessorTestSuite) TestProcessorWorkerProcessFailed() {
 	mockProcessor := mocks.NewDataProcessor[string](suite.T())
 	mockProcessor.EXPECT().Process("test").Return("", errors.New("test"))
 
-	worker := NewProcessorWorker(mockProcessor)
+	worker := newProcessorWorker(mockProcessor)
 	suite.NotNil(worker)
 
 	receiver := make(chan string, 1)
@@ -170,7 +170,7 @@ func (suite *proccessorTestSuite) TestProcessorWorkerProcessFailed() {
 }
 
 func (suite *proccessorTestSuite) TestNOOPProcessor() {
-	noop := NewNoOpProcessor()
+	noop := newNoOpProcessor()
 	suite.NotNil(noop)
 
 	data := &models.Token{