feat: generics support#8
Closed
thejoeejoee wants to merge 2 commits into
Closed
Conversation
thejoeejoee
force-pushed
the
feat-generics
branch
from
22fe43e to
c27e771
Compare
diff --git a/CHANGELOG.md b/CHANGELOG.md index c64304a..0aae4e3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). Note: This project is in early development. The API may change without warning in any 0.x version. +## [Unreleased] + +### Changed + +- Introduced golang generics support for entire library Batch, Source, and Processor interfaces +- Increased Go version requirement to 1.18 + ## [0.1.1] - 2024-07-18 ### Changed diff --git a/batch/batch.go b/batch/batch.go index 453aab7..4adbb1d 100644 --- a/batch/batch.go +++ b/batch/batch.go @@ -14,7 +14,7 @@ // to prioritize the parameters in some way. They are prioritized as follows // (with EOF signifying the end of the input data): // -// MaxTime = MaxItems > EOF > MinTime > MinItems +// MaxTime = MaxItems > EOF > MinTime > MinItems // // A few examples: // @@ -22,7 +22,7 @@ // processed right away. // // MinItems = 10, MinTime = 2s. After 1s, 10 items have been read. They are -// not processed until 2s has passed (along with all other items that have +// not processed until 2s has passed (along with all other in that have // been read up to the 2s mark). // // MaxItems = 10, MinTime = 2s. After 1s, 10 items have been read. They aren't @@ -48,10 +48,10 @@ import ( // To create a new Batch, call the New function. Creating one using &Batch{} // will return the default Batch. // -// // The following are equivalent -// defaultBatch1 := &batch.Batch{} -// defaultBatch2 := batch.New(nil) -// defaultBatch3 := batch.New(batch.NewConstantConfig(&batch.ConfigValues{})) +// // The following are equivalent +// defaultBatch1 := &batch.Batch{} +// defaultBatch2 := batch.New(nil) +// defaultBatch3 := batch.New(batch.NewConstantConfig(&batch.ConfigValues{})) // // The defaults (with nil Config) provide a usable, but likely suboptimal, Batch // where items are processed as soon as they are retrieved from the source. @@ -63,7 +63,7 @@ import ( // by Batch. For easier usage, the helper function NextItem can be used to // read from the input channel, set the data, and return the modified Item: // -// ps.Output() <- batch.NextItem(ps, item) +// ps.Output() <- batch.NextItem(ps, item) // // Batch runs asynchronously until the source closes its PipelineSource, signaling // that there is nothing else to read. Once that happens, and the pipeline has @@ -74,34 +74,36 @@ import ( // The first way can be used if errors need to be processed elsewhere. A simple // loop could look like this: // -// errs := myBatch.Go(ctx, s, p) -// for err := range errs { -// // Log the error here... -// log.Print(err.Error()) -// } -// // Now batch processing is done +// errs := myBatch.Go(ctx, s, p) +// for err := range errs { +// // Log the error here... +// log.Print(err.Error()) +// } +// // Now batch processing is done // // If the errors don't need to be processed, the IgnoreErrors function can be // used to drain the error channel. Then the Done channel can be used to // determine whether or not batch processing is complete: // -// batch.IgnoreErrors(myBatch.Go(ctx, s, p)) -// <-myBatch.Done() -// // Now batch processing is done +// batch.IgnoreErrors(myBatch.Go(ctx, s, p)) +// <-myBatch.Done() +// // Now batch processing is done // // Note that the errors returned on the error channel may be wrapped in a // batch.Error so the caller knows whether they come from the source or the // processor (or neither). Errors from the source will be of type SourceError, // and errors from the processor will be of type ProcessorError. Errors from // Batch itself will be neither. -type Batch struct { +type Batch[I any, O any] struct { config Config - src Source - proc Processor - items chan *Item - ids chan uint64 // For unique IDs - done chan struct{} + src Source[I, I] + proc Processor[I, O] + in chan *Item[I] + out chan *Item[O] + + ids chan uint64 // For unique IDs + done chan struct{} // mu protects the following variables. The reason errs is protected is // to avoid sending on a closed channel in the Go method. @@ -116,15 +118,15 @@ type Batch struct { // To avoid race conditions, the config cannot be changed after the Batch // is created. Instead, implement the Config interface to support changing // values. -func New(config Config) *Batch { - return &Batch{ +func New[I any, O any](config Config) *Batch[I, O] { + return &Batch[I, O]{ config: config, } } // Source reads items that are to be batch processed. -type Source interface { - // Read reads items from somewhere and writes them to the Output +type Source[I any, O any] interface { + // Read reads in from somewhere and writes them to the Output // channel of ps. Any errors it encounters while reading are written to the // Errors channel. The Input channel provides a steady stream of Items that // have pre-set metadata so the batch processor can identify them. A helper @@ -147,12 +149,12 @@ type Source interface { // } // // Read should not modify an item after adding it to items. - Read(ctx context.Context, ps *PipelineStage) + Read(ctx context.Context, ps *PipelineStage[I, O]) } // Processor processes items in batches. -type Processor interface { - // Process processes items from ps's Input channel and returns any errors +type Processor[I any, O any] interface { + // Process processes in from ps's Input channel and returns any errors // encountered on the Errors channel. When it is done, it must close ps // to signify that it's finished processing. Simply returning isn't enough. // @@ -189,7 +191,7 @@ type Processor interface { // Process may be run in any number of concurrent goroutines. If // concurrency needs to be limited it must be done in Process; for // example, by using a semaphore channel. - Process(ctx context.Context, ps *PipelineStage) + Process(ctx context.Context, ps *PipelineStage[I, O]) } // Go starts batch processing asynchronously and returns a channel on @@ -200,9 +202,9 @@ type Processor interface { // calls to Go are not allowed. If Go is called before a previous call // completes, the second one will panic. // -// // NOTE: bad - this will panic! -// errs := batch.Go(ctx, s, p) -// errs2 := batch.Go(ctx, s, p) // this call panics +// // NOTE: bad - this will panic! +// errs := batch.Go(ctx, s, p) +// errs2 := batch.Go(ctx, s, p) // this call panics // // Note that Go does not stop if ctx is done. Otherwise loss of data could occur. // Suppose the source reads item A and then ctx is canceled. If Go were to return @@ -215,7 +217,7 @@ type Processor interface { // done, it closes its error channel to signal to the batch processor. // Finally, the batch processor signals to its caller that processing is // complete and the entire pipeline is drained. -func (b *Batch) Go(ctx context.Context, s Source, p Processor) <-chan error { +func (b *Batch[I, O]) Go(ctx context.Context, s Source[I, I], p Processor[I, O]) <-chan error { b.mu.Lock() defer b.mu.Unlock() @@ -232,7 +234,8 @@ func (b *Batch) Go(ctx context.Context, s Source, p Processor) <-chan error { b.src = s b.proc = p - b.items = make(chan *Item) + b.in = make(chan *Item[I]) + b.out = make(chan *Item[O]) b.ids = make(chan uint64) b.done = make(chan struct{}) @@ -246,12 +249,12 @@ func (b *Batch) Go(ctx context.Context, s Source, p Processor) <-chan error { // Done provides an alternative way to determine when processing is // complete. When it is, the channel is closed, signaling that everything // is done. -func (b *Batch) Done() <-chan struct{} { +func (b *Batch[I, O]) Done() <-chan struct{} { return b.done } -// doIDGenerator generates unique IDs for the items in the pipeline. -func (b *Batch) doIDGenerator() { +// doIDGenerator generates unique IDs for the in in the pipeline. +func (b *Batch[I, O]) doIDGenerator() { for id := uint64(0); ; id++ { select { case b.ids <- id: @@ -263,11 +266,11 @@ func (b *Batch) doIDGenerator() { } // doReader starts the reader goroutine and reads from its channels. -func (b *Batch) doReader(ctx context.Context) { - in := make(chan *Item) - out := make(chan *Item) +func (b *Batch[I, O]) doReader(ctx context.Context) { + in := make(chan *Item[I]) + out := make(chan *Item[I]) errs := make(chan error) - ps := &PipelineStage{ + ps := &PipelineStage[I, I]{ Input: in, Output: out, Errors: errs, @@ -275,7 +278,7 @@ func (b *Batch) doReader(ctx context.Context) { go b.src.Read(ctx, ps) - nextItem := &Item{ + nextItem := &Item[I]{ id: <-b.ids, } @@ -283,13 +286,13 @@ func (b *Batch) doReader(ctx context.Context) { for !outClosed || !errClosed { select { case in <- nextItem: - nextItem = &Item{ + nextItem = &Item[I]{ id: <-b.ids, } case item, ok := <-out: if ok { - b.items <- item + b.in <- item } else { outClosed = true } @@ -305,11 +308,11 @@ func (b *Batch) doReader(ctx context.Context) { } } - close(b.items) + close(b.in) } // doProcessors starts the processor goroutine. -func (b *Batch) doProcessors(ctx context.Context) { +func (b *Batch[I, O]) doProcessors(ctx context.Context) { var wg sync.WaitGroup wg.Add(1) go func() { @@ -336,7 +339,7 @@ func fixConfig(c ConfigValues) ConfigValues { return c } -func (b *Batch) process(ctx context.Context) { +func (b *Batch[I, O]) process(ctx context.Context) { var ( wg sync.WaitGroup done bool @@ -356,10 +359,10 @@ func (b *Batch) process(ctx context.Context) { bufSize = 1024 } - var items = make([]*Item, 0, bufSize) + var items = make([]*Item[I], 0, bufSize) done, items = b.waitForItems(ctx, items, &config) - // TODO this resets the time whenever no items are available. Need to + // TODO this resets the time whenever no in are available. Need to // decide if that's the right way to do it. if len(items) == 0 { continue @@ -370,10 +373,10 @@ func (b *Batch) process(ctx context.Context) { go func() { defer wg.Done() - in := make(chan *Item) - out := make(chan *Item) + in := make(chan *Item[I]) + out := make(chan *Item[O]) errs := make(chan error) - ps := &PipelineStage{ + ps := &PipelineStage[I, O]{ Input: in, Output: out, Errors: errs, @@ -393,7 +396,7 @@ func (b *Batch) process(ctx context.Context) { select { case item, ok := <-out: if ok { - b.items <- item + b.out <- item } else { outClosed = true } @@ -418,7 +421,7 @@ func (b *Batch) process(ctx context.Context) { // waitForItems waits until enough items are read to begin batch processing, based // on config. It returns true if processing is completely finished, and false // otherwise. -func (b *Batch) waitForItems(ctx context.Context, items []*Item, config *ConfigValues) (bool, []*Item) { +func (b *Batch[I, O]) waitForItems(ctx context.Context, items []*Item[I], config *ConfigValues) (bool, []*Item[I]) { var ( reachedMinTime bool itemsRead uint64 @@ -445,7 +448,7 @@ func (b *Batch) waitForItems(ctx context.Context, items []*Item, config *ConfigV for { select { - case item, ok := <-b.items: + case item, ok := <-b.in: if ok { items = append(items, item) itemsRead++ diff --git a/batch/batch_test.go b/batch/batch_test.go index 8998386..ec69267 100644 --- a/batch/batch_test.go +++ b/batch/batch_test.go @@ -14,16 +14,16 @@ import ( ) type sourceFromSlice struct { - slice []interface{} + slice []int duration time.Duration } -func (s *sourceFromSlice) Read(ctx context.Context, ps *PipelineStage) { +func (s *sourceFromSlice) Read(ctx context.Context, ps *PipelineStage[int, int]) { defer ps.Close() for _, item := range s.slice { time.Sleep(s.duration) - ps.Output <- NextItem(ps, item) + ps.Output <- NextItem[int, int](ps, item) } } @@ -32,7 +32,7 @@ type processorCounter struct { num uint32 } -func (p *processorCounter) Process(ctx context.Context, ps *PipelineStage) { +func (p *processorCounter) Process(ctx context.Context, ps *PipelineStage[int, int]) { defer ps.Close() count := 0 @@ -62,11 +62,11 @@ func TestBatch_Go(t *testing.T) { t.Run("basic test", func(t *testing.T) { t.Parallel() - batch := &Batch{} + batch := &Batch[int, any]{} s := &sourceFromSlice{ - slice: []interface{}{1, 2, 3, 4, 5}, + slice: []int{1, 2, 3, 4, 5}, } - p := &processor.Nil{} + p := &processor.Nil[int, any]{} errs := batch.Go(context.Background(), s, p) @@ -86,11 +86,11 @@ func TestBatch_Go(t *testing.T) { t.Parallel() // Concurrent calls to Go should panic - batch := &Batch{} - s := &source.Nil{ + batch := &Batch[int, any]{} + s := &source.Nil[int]{ Duration: time.Second, } - p := &processor.Nil{ + p := &processor.Nil[int, any]{ Duration: 0, } @@ -116,11 +116,11 @@ func TestBatch_Go(t *testing.T) { t.Parallel() errSrc := errors.New("source") - batch := &Batch{} - s := &source.Error{ + batch := &Batch[int, any]{} + s := &source.Error[int]{ Err: errSrc, } - p := &processor.Nil{} + p := &processor.Nil[int, any]{} errs := batch.Go(context.Background(), s, p) @@ -146,11 +146,11 @@ func TestBatch_Go(t *testing.T) { t.Parallel() errProc := errors.New("processor") - batch := &Batch{} + batch := &Batch[int, any]{} s := &sourceFromSlice{ - slice: []interface{}{1}, + slice: []int{1}, } - p := &processor.Error{ + p := &processor.Error[int, any]{ Err: errProc, } @@ -297,12 +297,12 @@ func TestBatch_Go(t *testing.T) { t.Run(test.name, func(t *testing.T) { t.Parallel() - inputSlice := make([]interface{}, test.inputSize) + inputSlice := make([]int, test.inputSize) for i := 0; i < len(inputSlice); i++ { inputSlice[i] = rand.Int() } - batch := New(NewConstantConfig(test.config)) + batch := New[int, int](NewConstantConfig(test.config)) s := &sourceFromSlice{ slice: inputSlice, duration: test.inputDuration, @@ -326,11 +326,11 @@ func TestBatch_Done(t *testing.T) { t.Run("basic test", func(t *testing.T) { t.Parallel() - batch := &Batch{} - s := &source.Nil{ + batch := &Batch[int, any]{} + s := &source.Nil[int]{ Duration: 0, } - p := &processor.Nil{ + p := &processor.Nil[int, any]{ Duration: 0, } @@ -347,12 +347,12 @@ func TestBatch_Done(t *testing.T) { t.Run("with source sleep", func(t *testing.T) { t.Parallel() - batch := &Batch{} + batch := &Batch[int, any]{} s := &sourceFromSlice{ - slice: []interface{}{1}, + slice: []int{1}, duration: 100 * time.Millisecond, } - p := &processor.Nil{ + p := &processor.Nil[int, any]{ Duration: 10 * time.Millisecond, } @@ -372,12 +372,12 @@ func TestBatch_Done(t *testing.T) { t.Run("with processor sleep", func(t *testing.T) { t.Parallel() - batch := &Batch{} + batch := &Batch[int, any]{} s := &sourceFromSlice{ - slice: []interface{}{1}, + slice: []int{1}, duration: 10 * time.Millisecond, } - p := &processor.Nil{ + p := &processor.Nil[int, any]{ Duration: 100 * time.Millisecond, } diff --git a/batch/example_test.go b/batch/example_test.go index 9e9b4a9..4b10d46 100644 --- a/batch/example_test.go +++ b/batch/example_test.go @@ -13,17 +13,19 @@ import ( // printProcessor is a Processor that prints items in batches. // // To demonstrate how errors can be handled, it fails to process the number 5. -type printProcessor struct{} +type printProcessor[I, O any] struct{} // Process prints a batch of items. -func (p printProcessor) Process(ctx context.Context, ps *batch.PipelineStage) { +func (p printProcessor[I, O]) Process(ctx context.Context, ps *batch.PipelineStage[int, any]) { // Process needs to close ps after it's done defer ps.Close() - toPrint := make([]interface{}, 0, 5) + var errValue = 5 + + toPrint := make([]int, 0, 5) for item := range ps.Input { // Get returns the item itself - if item.Get() == 5 { + if item.Get() == errValue { ps.Errors <- errors.New("cannot process 5") continue } @@ -39,12 +41,12 @@ func Example() { config := batch.NewConstantConfig(&batch.ConfigValues{ MinItems: 5, }) - b := batch.New(config) - p := &printProcessor{} + b := batch.New[int, any](config) + p := &printProcessor[int, any]{} // Channel is a Source that reads from a channel until it's closed - ch := make(chan interface{}) - s := source.Channel{ + ch := make(chan int) + s := source.Channel[int]{ Input: ch, } diff --git a/batch/helpers.go b/batch/helpers.go index f57420e..b90db9a 100644 --- a/batch/helpers.go +++ b/batch/helpers.go @@ -4,12 +4,12 @@ package batch // its data, and returns it. If the input channel is closed, it returns nil. // NextItem can be used in the source Read function: // -// func (s *source) Read(ctx context.Context, ps batch.PipelineStage) { -// // Read data into myData... -// items <- batch.NextItem(ps, myData) -// // ... -// } -func NextItem(ps *PipelineStage, data interface{}) *Item { +// func (s *source) Read(ctx context.Context, ps batch.PipelineStage) { +// // Read data into myData... +// in <- batch.NextItem(ps, myData) +// // ... +// } +func NextItem[I, O any](ps *PipelineStage[I, O], data I) *Item[I] { i, ok := <-ps.Input if !ok { return nil @@ -22,12 +22,12 @@ func NextItem(ps *PipelineStage, data interface{}) *Item { // It can be used with Batch.Go if errors aren't needed. Since the error channel // is unbuffered, one cannot just throw away the error channel like this: // -// // NOTE: bad - this can cause a deadlock! -// _ = batch.Go(ctx, p, s) +// // NOTE: bad - this can cause a deadlock! +// _ = batch.Go(ctx, p, s) // // Instead, IgnoreErrors can be used to safely throw away all errors: // -// batch.IgnoreErrors(myBatch.Go(ctx, p, s)) +// batch.IgnoreErrors(myBatch.Go(ctx, p, s)) func IgnoreErrors(errs <-chan error) { // nil channels always block, so check for nil first to avoid a goroutine // leak diff --git a/batch/helpers_test.go b/batch/helpers_test.go index 34bc6ed..f306fb1 100644 --- a/batch/helpers_test.go +++ b/batch/helpers_test.go @@ -9,15 +9,15 @@ import ( func TestNextItem(t *testing.T) { t.Run("with item", func(t *testing.T) { - ch := make(chan *Item) + ch := make(chan *Item[any]) go func() { - ch <- &Item{ + ch <- &Item[any]{ id: 100, } }() - ps := &PipelineStage{ + ps := &PipelineStage[any, any]{ Input: ch, } @@ -31,10 +31,10 @@ func TestNextItem(t *testing.T) { }) t.Run("closed channel", func(t *testing.T) { - ch := make(chan *Item) + ch := make(chan *Item[any]) close(ch) - ps := &PipelineStage{ + ps := &PipelineStage[any, any]{ Input: ch, } diff --git a/batch/item.go b/batch/item.go index 7ea5f00..d4e0614 100644 --- a/batch/item.go +++ b/batch/item.go @@ -3,28 +3,28 @@ package batch import "sync" // Item holds a single item in the batch processing pipeline. -type Item struct { +type Item[T any] struct { mu sync.RWMutex id uint64 - item interface{} + item T } // GetID returns a unique ID of the current item in the pipeline. -func (i *Item) GetID() uint64 { +func (i *Item[T]) GetID() uint64 { i.mu.RLock() defer i.mu.RUnlock() return i.id } // Get returns the item data. -func (i *Item) Get() interface{} { +func (i *Item[T]) Get() T { i.mu.RLock() defer i.mu.RUnlock() return i.item } // Set sets the item data. -func (i *Item) Set(item interface{}) { +func (i *Item[T]) Set(item T) { i.mu.Lock() defer i.mu.Unlock() i.item = item diff --git a/batch/item_mock.go b/batch/item_mock.go index e760a6e..cb22eb3 100644 --- a/batch/item_mock.go +++ b/batch/item_mock.go @@ -8,7 +8,7 @@ import "sync" type MockItemGenerator struct { closeOnce sync.Once done chan struct{} - ch chan *Item + ch chan *Item[int] } // NewMockItemGenerator returns a new MockItemGenerator. @@ -17,19 +17,19 @@ type MockItemGenerator struct { func NewMockItemGenerator() *MockItemGenerator { m := &MockItemGenerator{ done: make(chan struct{}), - ch: make(chan *Item), + ch: make(chan *Item[int]), } go func() { id := uint64(0) - nextItem := &Item{ + nextItem := &Item[int]{ id: id, } for { select { case m.ch <- nextItem: id++ - nextItem = &Item{ + nextItem = &Item[int]{ id: id, } @@ -50,6 +50,6 @@ func (m *MockItemGenerator) Close() { } // GetCh returns a channel of Items with unique IDs. -func (m *MockItemGenerator) GetCh() <-chan *Item { +func (m *MockItemGenerator) GetCh() <-chan *Item[int] { return m.ch } diff --git a/batch/pipeline_stage.go b/batch/pipeline_stage.go index 228c5a7..296c2ba 100644 --- a/batch/pipeline_stage.go +++ b/batch/pipeline_stage.go @@ -2,12 +2,12 @@ package batch // PipelineStage contains the input and output channels for a single // stage of the batch pipeline. -type PipelineStage struct { - // Input contains the input items for a pipeline stage. - Input <-chan *Item +type PipelineStage[I any, O any] struct { + // Input contains the input in for a pipeline stage. + Input <-chan *Item[I] // Output is for the output of the pipeline stage. - Output chan<- *Item + Output chan<- *Item[O] //Retry chan<- *Item @@ -19,7 +19,7 @@ type PipelineStage struct { // // Note that it will also close the write channels. Do not close them separately // or it will panic. -func (p *PipelineStage) Close() { +func (p *PipelineStage[T, O]) Close() { close(p.Output) close(p.Errors) } diff --git a/batch/pipeline_stage_test.go b/batch/pipeline_stage_test.go index 0740e30..1445247 100644 --- a/batch/pipeline_stage_test.go +++ b/batch/pipeline_stage_test.go @@ -3,9 +3,9 @@ package batch import "testing" func TestPipelineStage_Close(t *testing.T) { - out := make(chan *Item) + out := make(chan *Item[any]) errs := make(chan error) - ps := &PipelineStage{ + ps := &PipelineStage[any, any]{ Output: out, Errors: errs, } diff --git a/go.mod b/go.mod index 7e85fd7..f9272dd 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,3 @@ module github.com/MasterOfBinary/gobatch -go 1.11 +go 1.18 diff --git a/processor/error.go b/processor/error.go index 273b8e9..2955d5e 100644 --- a/processor/error.go +++ b/processor/error.go @@ -7,12 +7,12 @@ import ( ) // Error returns a Processor that returns an error while processing. -type Error struct { +type Error[I, O any] struct { Err error } // Process discards all data sent to it after a certain amount of time. -func (p *Error) Process(ctx context.Context, ps *batch.PipelineStage) { +func (p *Error[I, O]) Process(ctx context.Context, ps *batch.PipelineStage[I, O]) { ps.Errors <- p.Err ps.Close() } diff --git a/processor/nil.go b/processor/nil.go index ef6b66e..cb8265f 100644 --- a/processor/nil.go +++ b/processor/nil.go @@ -9,12 +9,12 @@ import ( // Nil is a Processor that discards all data after a specified duration. // It can be used as a mock Processor. -type Nil struct { +type Nil[I, O any] struct { Duration time.Duration } // Process discards all data sent to it after a certain amount of time. -func (p *Nil) Process(ctx context.Context, ps *batch.PipelineStage) { +func (p *Nil[I, O]) Process(ctx context.Context, ps *batch.PipelineStage[I, O]) { time.Sleep(p.Duration) ps.Close() } diff --git a/source/channel.go b/source/channel.go index 53cd536..bffb06a 100644 --- a/source/channel.go +++ b/source/channel.go @@ -8,12 +8,12 @@ import ( // Channel is a Source that reads from input until it is closed. // The input channel can be buffered or unbuffered. -type Channel struct { - Input <-chan interface{} +type Channel[I any] struct { + Input <-chan I } // Read reads from items until the input channel is closed. -func (s *Channel) Read(ctx context.Context, ps* batch.PipelineStage) { +func (s *Channel[I]) Read(ctx context.Context, ps *batch.PipelineStage[I, I]) { defer ps.Close() out := ps.Output diff --git a/source/channel_test.go b/source/channel_test.go index 505ab15..6179fb4 100644 --- a/source/channel_test.go +++ b/source/channel_test.go @@ -18,14 +18,14 @@ func TestChannelSource_Read(t *testing.T) { var wg sync.WaitGroup defer wg.Wait() - in := make(chan interface{}, size) - items := make(chan *batch.Item) + in := make(chan any, size) + items := make(chan *batch.Item[any]) _ = make(chan error) itemGen := batch.NewMockItemGenerator() defer itemGen.Close() - s := Channel{ + s := Channel[any]{ Input: in, } diff --git a/source/error.go b/source/error.go index de00bc9..8ab36e4 100644 --- a/source/error.go +++ b/source/error.go @@ -8,12 +8,12 @@ import ( // Error is a Source that returns an error and then closes immediately. // It can be used as a mock Source. -type Error struct { +type Error[I any] struct { Err error } // Read returns an error and then closes. -func (s *Error) Read(ctx context.Context, ps *batch.PipelineStage) { +func (s *Error[I]) Read(ctx context.Context, ps *batch.PipelineStage[I, I]) { ps.Errors <- s.Err ps.Close() } diff --git a/source/nil.go b/source/nil.go index cdf85c0..23ad5e4 100644 --- a/source/nil.go +++ b/source/nil.go @@ -10,12 +10,12 @@ import ( // Nil is a Source that doesn't read any data. Instead it closes the // pipeline stage after specified duration. It can be used as a mock // Source. -type Nil struct { +type Nil[I any] struct { Duration time.Duration } // Read doesn't read anything. -func (s *Nil) Read(ctx context.Context, ps *batch.PipelineStage) { +func (s *Nil[I]) Read(ctx context.Context, ps *batch.PipelineStage[I, I]) { time.Sleep(s.Duration) ps.Close() }
diff --git a/batch/batch.go b/batch/batch.go index 4adbb1d..675959d 100644 --- a/batch/batch.go +++ b/batch/batch.go @@ -421,7 +421,7 @@ func (b *Batch[I, O]) process(ctx context.Context) { // waitForItems waits until enough items are read to begin batch processing, based // on config. It returns true if processing is completely finished, and false // otherwise. -func (b *Batch[I, O]) waitForItems(ctx context.Context, items []*Item[I], config *ConfigValues) (bool, []*Item[I]) { +func (b *Batch[I, O]) waitForItems(_ context.Context, items []*Item[I], config *ConfigValues) (bool, []*Item[I]) { var ( reachedMinTime bool itemsRead uint64 diff --git a/batch/batch_test.go b/batch/batch_test.go index ec69267..2af1c91 100644 --- a/batch/batch_test.go +++ b/batch/batch_test.go @@ -18,7 +18,7 @@ type sourceFromSlice struct { duration time.Duration } -func (s *sourceFromSlice) Read(ctx context.Context, ps *PipelineStage[int, int]) { +func (s *sourceFromSlice) Read(_ context.Context, ps *PipelineStage[int, int]) { defer ps.Close() for _, item := range s.slice { @@ -32,7 +32,7 @@ type processorCounter struct { num uint32 } -func (p *processorCounter) Process(ctx context.Context, ps *PipelineStage[int, int]) { +func (p *processorCounter) Process(_ context.Context, ps *PipelineStage[int, int]) { defer ps.Close() count := 0 @@ -126,14 +126,13 @@ func TestBatch_Go(t *testing.T) { var found bool for err := range errs { - if src, ok := err.(*SourceError); ok { - if src.Original() == errSrc { + var src *SourceError + if errors.As(err, &src) { + if errors.Is(errSrc, src.Original()) { found = true } else { t.Errorf("Found source error %v, want %v", src.Original(), errSrc) } - } else { - t.Error("Found an unexpected error") } } @@ -158,14 +157,13 @@ func TestBatch_Go(t *testing.T) { var found bool for err := range errs { - if proc, ok := err.(*ProcessorError); ok { - if proc.Original() == errProc { + var proc *ProcessorError + if errors.As(err, &proc) { + if errors.Is(errProc, proc.Original()) { found = true } else { t.Errorf("Found processor error %v, want %v", proc.Original(), errProc) } - } else { - t.Error("Found an unexpected error") } } diff --git a/batch/example_test.go b/batch/example_test.go index 4b10d46..721445d 100644 --- a/batch/example_test.go +++ b/batch/example_test.go @@ -16,7 +16,7 @@ import ( type printProcessor[I, O any] struct{} // Process prints a batch of items. -func (p printProcessor[I, O]) Process(ctx context.Context, ps *batch.PipelineStage[int, any]) { +func (p printProcessor[I, O]) Process(_ context.Context, ps *batch.PipelineStage[int, any]) { // Process needs to close ps after it's done defer ps.Close() diff --git a/processor/error.go b/processor/error.go index 2955d5e..e6477dd 100644 --- a/processor/error.go +++ b/processor/error.go @@ -12,7 +12,7 @@ type Error[I, O any] struct { } // Process discards all data sent to it after a certain amount of time. -func (p *Error[I, O]) Process(ctx context.Context, ps *batch.PipelineStage[I, O]) { +func (p *Error[I, O]) Process(_ context.Context, ps *batch.PipelineStage[I, O]) { ps.Errors <- p.Err ps.Close() } diff --git a/processor/nil.go b/processor/nil.go index cb8265f..9e71433 100644 --- a/processor/nil.go +++ b/processor/nil.go @@ -14,7 +14,7 @@ type Nil[I, O any] struct { } // Process discards all data sent to it after a certain amount of time. -func (p *Nil[I, O]) Process(ctx context.Context, ps *batch.PipelineStage[I, O]) { +func (p *Nil[I, O]) Process(_ context.Context, ps *batch.PipelineStage[I, O]) { time.Sleep(p.Duration) ps.Close() } diff --git a/source/channel.go b/source/channel.go index bffb06a..8fa51e9 100644 --- a/source/channel.go +++ b/source/channel.go @@ -13,7 +13,7 @@ type Channel[I any] struct { } // Read reads from items until the input channel is closed. -func (s *Channel[I]) Read(ctx context.Context, ps *batch.PipelineStage[I, I]) { +func (s *Channel[I]) Read(_ context.Context, ps *batch.PipelineStage[I, I]) { defer ps.Close() out := ps.Output diff --git a/source/error.go b/source/error.go index 8ab36e4..56832eb 100644 --- a/source/error.go +++ b/source/error.go @@ -13,7 +13,7 @@ type Error[I any] struct { } // Read returns an error and then closes. -func (s *Error[I]) Read(ctx context.Context, ps *batch.PipelineStage[I, I]) { +func (s *Error[I]) Read(_ context.Context, ps *batch.PipelineStage[I, I]) { ps.Errors <- s.Err ps.Close() } diff --git a/source/nil.go b/source/nil.go index 23ad5e4..4152640 100644 --- a/source/nil.go +++ b/source/nil.go @@ -15,7 +15,7 @@ type Nil[I any] struct { } // Read doesn't read anything. -func (s *Nil[I]) Read(ctx context.Context, ps *batch.PipelineStage[I, I]) { +func (s *Nil[I]) Read(_ context.Context, ps *batch.PipelineStage[I, I]) { time.Sleep(s.Duration) ps.Close() }
thejoeejoee
force-pushed
the
feat-generics
branch
from
c27e771 to
7737335
Compare
Owner
|
Closing as superseded by merged generics support in v0.5.0. Thanks for the early generics work and proposal. |
MasterOfBinary
added a commit
that referenced
this pull request
May 29, 2026
Fix #8: the Nil processor was described as "passes items through unchanged (for benchmarking)", which is wrong. Per processor/nil.go it sleeps for a configurable Duration (and can mark items cancelled via MarkCancelled). Corrected in CLAUDE.md, AGENTS.md, and README.md to match processor/doc.go. Fix #9: the engine emits wrapped errors as pointers (&SourceError{}, &ProcessorError{} in batch/batch.go), so errors.As must target the pointer type. Made the CLAUDE.md/AGENTS.md guidance explicit with a concrete example; README already used the pointer form. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
introduced support for native golang generics