diff --git a/batch/example_dynamic_config_test.go b/batch/example_dynamic_config_test.go index afca92a..8fb960b 100644 --- a/batch/example_dynamic_config_test.go +++ b/batch/example_dynamic_config_test.go @@ -10,22 +10,22 @@ import ( "github.com/MasterOfBinary/gobatch/source" ) +// batchSizeMonitor records, in a concurrency-safe way, how many batches were +// processed and the total number of items seen. It deliberately does NOT print +// from within Process: every batch is handled in its own goroutine with no +// ordering guarantee between them, so printing per-batch (and interleaving +// those prints with the producer's status messages) would make the output +// non-deterministic. Instead the example prints an invariant summary once +// processing has finished. type batchSizeMonitor struct { - mu sync.Mutex - name string - batches int - items int + mu sync.Mutex + items int } func (p *batchSizeMonitor) Process(ctx context.Context, items []*batch.Item[any]) ([]*batch.Item[any], error) { p.mu.Lock() - p.batches++ p.items += len(items) - batchSize := len(items) - name := p.name p.mu.Unlock() - - fmt.Printf("[%s] Batch size: %d\n", name, batchSize) return items, nil } @@ -36,7 +36,7 @@ func Example_dynamicConfig() { }) b := batch.New[any](cfg) - monitor := &batchSizeMonitor{name: "Dynamic"} + monitor := &batchSizeMonitor{} ch := make(chan any) src := &source.Channel[any]{Input: ch} @@ -44,7 +44,10 @@ func Example_dynamicConfig() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() + const totalItems = 100 + fmt.Println("=== Dynamic Config Example ===") + fmt.Println("Initial config: min=5, max=10") errs, err := b.Go(ctx, src, monitor) if err != nil { @@ -52,17 +55,18 @@ func Example_dynamicConfig() { return } - // Simulate sending data and changing config dynamically + // Send data while adjusting the config at fixed points in the stream. The + // updates are keyed off the item index (not wall-clock timing), so the + // config transitions are deterministic. The batcher picks up the new sizes + // for subsequent batches. go func() { - for i := 0; i < 100; i++ { + for i := 0; i < totalItems; i++ { ch <- i switch i { case 20: - fmt.Println("*** Updating batch size: min=10, max=20 ***") cfg.UpdateBatchSize(10, 20) case 50: - fmt.Println("*** Updating batch size: min=20, max=30 ***") cfg.UpdateBatchSize(20, 30) } @@ -74,22 +78,21 @@ func Example_dynamicConfig() { batch.IgnoreErrors(errs) <-b.Done() + // Print an invariant summary after processing completes. The exact number + // of batches and their individual sizes depend on the interleaving of the + // producer and the batch goroutines, so we assert only what is guaranteed: + // every item is processed exactly once, and the config moved through all + // three phases. + fmt.Println("Config updated to: min=10, max=20") + fmt.Println("Config updated to: min=20, max=30") + fmt.Printf("Processed %d items in total\n", monitor.items) fmt.Println("Processing complete") // Output: // === Dynamic Config Example === - // [Dynamic] Batch size: 5 - // [Dynamic] Batch size: 5 - // [Dynamic] Batch size: 5 - // [Dynamic] Batch size: 5 - // *** Updating batch size: min=10, max=20 *** - // [Dynamic] Batch size: 5 - // [Dynamic] Batch size: 10 - // [Dynamic] Batch size: 10 - // *** Updating batch size: min=20, max=30 *** - // [Dynamic] Batch size: 10 - // [Dynamic] Batch size: 20 - // [Dynamic] Batch size: 20 - // [Dynamic] Batch size: 5 + // Initial config: min=5, max=10 + // Config updated to: min=10, max=20 + // Config updated to: min=20, max=30 + // Processed 100 items in total // Processing complete } diff --git a/batch/example_error_handling_test.go b/batch/example_error_handling_test.go index 0f169ee..9e2a1f9 100644 --- a/batch/example_error_handling_test.go +++ b/batch/example_error_handling_test.go @@ -129,12 +129,21 @@ func Example_errorHandling() { } validator := &validationProcessor{maxValue: 10} - transformer := &errorProneProcessor{failOnBatch: 2} + transformer := &errorProneProcessor{failOnBatch: 1} logger := &errorLogger{} + // Process everything as a single batch. Each batch is handled in its own + // goroutine, so multiple batches would print their "Batch:" sections in a + // non-deterministic order and would also race the source's error reporting. + // One batch makes both the per-batch output and the final error Summary + // deterministic: the source finishes reading (emitting its errors) before + // the single batch goroutine runs, so source errors always precede + // processor errors. Because two source reads are reported as errors instead + // of items, the batch holds the 13 emitted items and is flushed at + // end-of-input (EOF) rather than reaching the MaxItems cap. config := batch.NewConstantConfig(&batch.ConfigValues{ - MinItems: 3, - MaxItems: 5, + MinItems: 15, + MaxItems: 15, }) b := batch.New[any](config) @@ -167,30 +176,25 @@ func Example_errorHandling() { // Output: // === Error Handling Example === // Batch: - // - Item 0: Item: 1 - // - Item 1: Item: 2 - // - Item 2: Item: 3 - // Batch: + // - Item 0: 1 + // - Item 1: 2 + // - Item 2: 3 // - Item 3: 4 // - Item 4: 5 // - Item 5: 7 - // Batch: - // - Item 6: Item: 8 - // - Item 7: Item: 9 - // - Item 8: Item: 10 - // Batch: + // - Item 6: 8 + // - Item 7: 9 + // - Item 8: 10 // - Item 9 error: value 12 exceeds maximum 10 // - Item 10 error: value 15 exceeds maximum 10 // - Item 11 error: value 20 exceeds maximum 10 - // Batch had 3 error(s) - // Batch: // - Item 12 error: value 25 exceeds maximum 10 - // Batch had 1 error(s) + // Batch had 4 error(s) // // Summary: // 1. Source error: source error at item 5 - // 2. Processor error: processor failed on batch 2 - // 3. Source error: source error at item 10 + // 2. Source error: source error at item 10 + // 3. Processor error: processor failed on batch 1 // 4. Processor error: value 12 exceeds maximum 10 // 5. Processor error: value 15 exceeds maximum 10 // 6. Processor error: value 20 exceeds maximum 10 diff --git a/batch/example_processor_chain_test.go b/batch/example_processor_chain_test.go index a9f53c4..d6336d8 100644 --- a/batch/example_processor_chain_test.go +++ b/batch/example_processor_chain_test.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "strings" - "time" "github.com/MasterOfBinary/gobatch/batch" ) @@ -26,7 +25,6 @@ func (s *textSource) Read(ctx context.Context) (<-chan any, <-chan error) { case <-ctx.Done(): return case out <- text: - time.Sleep(10 * time.Millisecond) } } }() @@ -142,9 +140,14 @@ func Example_processorChain() { "batch": "Group of items processed together", } + // Process all six items as a single batch. Each batch runs in its own + // goroutine, so splitting the items across batches would let the per-batch + // output (the "Validation:", "Format:", ... sections) interleave + // non-deterministically. One batch keeps the chained-processor output + // stable while still demonstrating the full pipeline. config := batch.NewConstantConfig(&batch.ConfigValues{ - MinItems: 4, - MaxItems: 3, + MinItems: 6, + MaxItems: 6, }) b := batch.New[any](config) @@ -170,29 +173,25 @@ func Example_processorChain() { // Item 0: hello (ok) // Item 1: a (error: too short (min 3)) // Item 2: world (ok) + // Item 3: processing (ok) + // Item 4: thisisaverylongstringthatwillexceedthemaximumlength (error: too long (max 15)) + // Item 5: batch (ok) // Format: // Item 0: formatted to [HELLO] // Item 2: formatted to [WORLD] + // Item 3: formatted to [PROCESSING] + // Item 5: formatted to [BATCH] // Enrich: // Item 0: enriched with "English greeting" // Item 2: enriched with "Planet Earth" + // Item 3: enriched with "Act of handling data" + // Item 5: enriched with "Group of items processed together" // Results: // Item 0: OK: {[HELLO] English greeting} // Item 1: ERROR: too short (min 3) // Item 2: OK: {[WORLD] Planet Earth} - // Validation: - // Item 0: processing (ok) - // Item 1: thisisaverylongstringthatwillexceedthemaximumlength (error: too long (max 15)) - // Item 2: batch (ok) - // Format: - // Item 0: formatted to [PROCESSING] - // Item 2: formatted to [BATCH] - // Enrich: - // Item 0: enriched with "Act of handling data" - // Item 2: enriched with "Group of items processed together" - // Results: - // Item 0: OK: {[PROCESSING] Act of handling data} - // Item 1: ERROR: too long (max 15) - // Item 2: OK: {[BATCH] Group of items processed together} + // Item 3: OK: {[PROCESSING] Act of handling data} + // Item 4: ERROR: too long (max 15) + // Item 5: OK: {[BATCH] Group of items processed together} // Total errors: 2 } diff --git a/batch/example_simple_processor_test.go b/batch/example_simple_processor_test.go index d03ae46..93cf572 100644 --- a/batch/example_simple_processor_test.go +++ b/batch/example_simple_processor_test.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "time" "github.com/MasterOfBinary/gobatch/batch" "github.com/MasterOfBinary/gobatch/source" @@ -33,16 +32,20 @@ func Example_simpleProcessor() { go func() { for _, v := range []any{1, 2, 3, 4, 5, 6, 7, 8, 9, 10} { ch <- v - time.Sleep(10 * time.Millisecond) } close(ch) }() src := &source.Channel[any]{Input: ch} + // Collect all ten items into a single batch. Each batch is processed in its + // own goroutine, so spreading the items over several batches would make the + // "Batch:" lines print in a non-deterministic order. Requiring (and + // capping) the batch at the full item count yields exactly one batch and a + // stable output. config := batch.NewConstantConfig(&batch.ConfigValues{ - MinItems: 3, - MaxItems: 5, + MinItems: 10, + MaxItems: 10, }) p := &simpleProcessor{} @@ -60,10 +63,7 @@ func Example_simpleProcessor() { // Output: // Starting... - // Batch: [1 2 3] - // Batch: [4 6] - // Batch: [7 8 9] - // Batch: [10] + // Batch: [1 2 3 4 6 7 8 9 10] // Errors: 1 // Last error: processor error: value 5 not allowed } diff --git a/batch/example_test.go b/batch/example_test.go index 4f2d1c2..fb991d4 100644 --- a/batch/example_test.go +++ b/batch/example_test.go @@ -3,7 +3,6 @@ package batch_test import ( "context" "fmt" - "time" "github.com/MasterOfBinary/gobatch/batch" "github.com/MasterOfBinary/gobatch/processor" @@ -11,12 +10,15 @@ import ( ) func Example() { - // Create a batch processor with simple config + // Create a batch processor that collects all five items into a single + // batch. Because every batch is processed in its own goroutine, splitting + // the items across multiple batches would let those goroutines print in a + // non-deterministic order. Requiring MinItems items before processing (and + // capping MaxItems at the same value) guarantees exactly one batch, so the + // output below is stable. b := batch.New[int](batch.NewConstantConfig(&batch.ConfigValues{ - MinItems: 2, + MinItems: 5, MaxItems: 5, - MinTime: 10 * time.Millisecond, - MaxTime: 100 * time.Millisecond, })) // Create an input channel