test: make Example_* output deterministic

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,33 +36,37 @@ func Example_dynamicConfig() {
 	})
 
 	b := batch.New[any](cfg)
-	monitor := &batchSizeMonitor{name: "Dynamic"}
+	monitor := &batchSizeMonitor{}
 
 	ch := make(chan any)
 	src := &source.Channel[any]{Input: ch}
 
 	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 {
 		fmt.Println(err)
 		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
 }

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

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
 }

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
 }

batch/example_test.go

@@ -3,20 +3,22 @@ package batch_test
 import (
 	"context"
 	"fmt"
-	"time"
 
 	"github.com/MasterOfBinary/gobatch/batch"
 	"github.com/MasterOfBinary/gobatch/processor"
 	"github.com/MasterOfBinary/gobatch/source"
 )
 
 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