From 02123df8cc15896f97c1580e49d8e2a991e37ddd Mon Sep 17 00:00:00 2001 From: sangtran Date: Thu, 3 Sep 2026 23:10:33 +0700 Subject: Moving all legacy quacker source code from github to cgit --- pipeline/batch.go | 64 +++++++++ pipeline/batch_test.go | 106 ++++++++++++++ pipeline/example_test.go | 53 +++++++ pipeline/options.go | 34 +++++ pipeline/pipeline.go | 21 +++ pipeline/pipeline_bench_test.go | 55 +++++++ pipeline/pipeline_test.go | 255 +++++++++++++++++++++++++++++++++ pipeline/stage.go | 310 ++++++++++++++++++++++++++++++++++++++++ pipeline/stage_test.go | 308 +++++++++++++++++++++++++++++++++++++++ 9 files changed, 1206 insertions(+) create mode 100644 pipeline/batch.go create mode 100644 pipeline/batch_test.go create mode 100644 pipeline/example_test.go create mode 100644 pipeline/options.go create mode 100644 pipeline/pipeline.go create mode 100644 pipeline/pipeline_bench_test.go create mode 100644 pipeline/pipeline_test.go create mode 100644 pipeline/stage.go create mode 100644 pipeline/stage_test.go (limited to 'pipeline') diff --git a/pipeline/batch.go b/pipeline/batch.go new file mode 100644 index 0000000..74b08f0 --- /dev/null +++ b/pipeline/batch.go @@ -0,0 +1,64 @@ +package pipeline + +import ( + "context" + "time" +) + +// Batch collects items into slices of up to size, flushing when full or when timeout elapses. +func Batch[T any](ctx context.Context, in <-chan T, size int, timeout time.Duration) <-chan []T { + if in == nil { + panic("pipeline: Batch input channel must not be nil") + } + if size < 1 { + panic("pipeline: Batch size must be at least 1") + } + + out := make(chan []T) + + go func() { + defer close(out) + + buf := make([]T, 0, size) + ticker := time.NewTicker(timeout) + defer ticker.Stop() + + flush := func() bool { + if len(buf) == 0 { + return true + } + batch := buf + buf = make([]T, 0, size) + select { + case out <- batch: + return true + case <-ctx.Done(): + return false + } + } + + for { + select { + case v, ok := <-in: + if !ok { + flush() + return + } + buf = append(buf, v) + if len(buf) >= size { + if !flush() { + return + } + } + case <-ticker.C: + if !flush() { + return + } + case <-ctx.Done(): + return + } + } + }() + + return out +} diff --git a/pipeline/batch_test.go b/pipeline/batch_test.go new file mode 100644 index 0000000..4d79ddc --- /dev/null +++ b/pipeline/batch_test.go @@ -0,0 +1,106 @@ +package pipeline + +import ( + "context" + "testing" + "time" +) + +func TestBatch_Full(t *testing.T) { + t.Parallel() + + input := make(chan int, 10) + for i := 1; i <= 10; i++ { + input <- i + } + close(input) + + out := Batch(t.Context(), input, 3, time.Second) + + var batches [][]int + for b := range out { + batches = append(batches, b) + } + + if len(batches) != 4 { + t.Fatalf("expected 4 batches, got %d: %v", len(batches), batches) + } + if len(batches[0]) != 3 { + t.Fatalf("expected first batch size 3, got %d", len(batches[0])) + } + if len(batches[3]) != 1 { + t.Fatalf("expected last batch size 1, got %d", len(batches[3])) + } +} + +func TestBatch_TimeoutFlush(t *testing.T) { + t.Parallel() + + input := make(chan int) + + out := Batch(t.Context(), input, 100, 50*time.Millisecond) + + input <- 1 + input <- 2 + + select { + case batch := <-out: + if len(batch) != 2 { + t.Fatalf("expected batch of 2 from timeout flush, got %d", len(batch)) + } + case <-time.After(time.Second): + t.Fatal("timeout waiting for batch flush") + } + + close(input) + for range out { + } +} + +func TestBatch_ContextCancel(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + + input := make(chan int) + out := Batch(ctx, input, 100, time.Second) + + cancel() + + done := make(chan struct{}) + go func() { + for range out { + } + close(done) + }() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("timeout waiting for shutdown") + } +} + +func TestBatch_NilInput(t *testing.T) { + t.Parallel() + + defer func() { + if r := recover(); r == nil { + t.Error("expected panic for nil input") + } + }() + + Batch[int](t.Context(), nil, 10, time.Second) +} + +func TestBatch_PanicOnZeroSize(t *testing.T) { + t.Parallel() + + defer func() { + if r := recover(); r == nil { + t.Error("expected panic for batch size < 1") + } + }() + + input := make(chan int) + Batch(t.Context(), input, 0, time.Second) +} diff --git a/pipeline/example_test.go b/pipeline/example_test.go new file mode 100644 index 0000000..e322529 --- /dev/null +++ b/pipeline/example_test.go @@ -0,0 +1,53 @@ +package pipeline_test + +import ( + "context" + "fmt" + "sort" + + "git.sangtran.io.vn/quacker.git/pipeline" +) + +func ExampleRun() { + input := make(chan int, 5) + for i := 1; i <= 5; i++ { + input <- i + } + close(input) + + out := pipeline.Run(context.Background(), input, + pipeline.Filter(func(v int) bool { return v%2 == 0 }), + pipeline.Map(1, func(_ context.Context, v int) (int, bool) { + return v * 10, true + }), + ) + + var results []int + for v := range out { + results = append(results, v) + } + sort.Ints(results) + fmt.Println(results) + // Output: [20 40] +} + +func ExampleSink() { + input := make(chan string, 3) + input <- "hello" + input <- "world" + input <- "!" + close(input) + + var collected []string + out := pipeline.Run(context.Background(), input, + pipeline.Sink(1, func(_ context.Context, v string) { + collected = append(collected, v) + }), + ) + + for range out { + } + + fmt.Println(collected) + // Output: [hello world !] +} diff --git a/pipeline/options.go b/pipeline/options.go new file mode 100644 index 0000000..c8c616e --- /dev/null +++ b/pipeline/options.go @@ -0,0 +1,34 @@ +package pipeline + +// StageObserver receives callbacks for stage lifecycle events. +type StageObserver interface { + OnItem() + OnDrop() + OnDone() +} + +// StageOption configures a stage's behavior. +type StageOption func(*stageConfig) + +type stageConfig struct { + observer StageObserver + bufferSize int +} + +// WithObserver attaches an observer to a stage. +func WithObserver(obs StageObserver) StageOption { + return func(cfg *stageConfig) { cfg.observer = obs } +} + +// WithBufferSize sets the output channel buffer size for a stage. +func WithBufferSize(size int) StageOption { + return func(cfg *stageConfig) { cfg.bufferSize = size } +} + +func applyOpts(opts []StageOption) stageConfig { + var cfg stageConfig + for _, opt := range opts { + opt(&cfg) + } + return cfg +} diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go new file mode 100644 index 0000000..170a04e --- /dev/null +++ b/pipeline/pipeline.go @@ -0,0 +1,21 @@ +// Package pipeline provides composable, type-safe stream processing with backpressure. +package pipeline + +import "context" + +// Stage transforms an input channel into an output channel. +// The stage owns the output channel and closes it when the input +// is exhausted or the context is cancelled. +type Stage[T any] func(ctx context.Context, in <-chan T) <-chan T + +// Run connects stages into a pipeline and returns the final output channel. +func Run[T any](ctx context.Context, input <-chan T, stages ...Stage[T]) <-chan T { + if input == nil { + panic("pipeline: input channel must not be nil") + } + ch := input + for _, stage := range stages { + ch = stage(ctx, ch) + } + return ch +} diff --git a/pipeline/pipeline_bench_test.go b/pipeline/pipeline_bench_test.go new file mode 100644 index 0000000..f4f81a0 --- /dev/null +++ b/pipeline/pipeline_bench_test.go @@ -0,0 +1,55 @@ +package pipeline + +import ( + "context" + "testing" +) + +func BenchmarkMap(b *testing.B) { + b.ReportAllocs() + + input := make(chan int, 256) + + out := Run(b.Context(), input, + Map(4, func(_ context.Context, v int) (int, bool) { + return v * 2, true + }), + ) + + go func() { + for range out { + } + }() + + for b.Loop() { + input <- 1 + } + + close(input) +} + +func BenchmarkPipeline_Composition(b *testing.B) { + b.ReportAllocs() + + input := make(chan int, 256) + + out := Run(b.Context(), input, + Filter(func(v int) bool { return v > 0 }), + Map(4, func(_ context.Context, v int) (int, bool) { + return v * 2, true + }), + Buffer[int](64), + Sink(2, func(_ context.Context, v int) {}), + ) + + go func() { + for range out { + } + }() + + for b.Loop() { + input <- 1 + } + + close(input) +} diff --git a/pipeline/pipeline_test.go b/pipeline/pipeline_test.go new file mode 100644 index 0000000..9593356 --- /dev/null +++ b/pipeline/pipeline_test.go @@ -0,0 +1,255 @@ +package pipeline + +import ( + "context" + "fmt" + "sync/atomic" + "testing" + "time" +) + +func TestRun_Passthrough(t *testing.T) { + t.Parallel() + + input := make(chan int, 3) + input <- 1 + input <- 2 + input <- 3 + close(input) + + out := Run(t.Context(), input) + + var got []int + for v := range out { + got = append(got, v) + } + + if len(got) != 3 { + t.Fatalf("expected 3 items, got %d", len(got)) + } +} + +func TestRun_NilInput(t *testing.T) { + t.Parallel() + + defer func() { + if r := recover(); r == nil { + t.Error("expected panic for nil input") + } + }() + + Run[int](t.Context(), nil) +} + +func TestRun_Composition(t *testing.T) { + t.Parallel() + + input := make(chan int, 10) + for i := 1; i <= 10; i++ { + input <- i + } + close(input) + + out := Run(t.Context(), input, + Filter(func(v int) bool { return v%2 == 0 }), + Map(1, func(_ context.Context, v int) (int, bool) { + return v * 2, true + }), + ) + + var got []int + for v := range out { + got = append(got, v) + } + + if len(got) != 5 { + t.Fatalf("expected 5 items, got %d: %v", len(got), got) + } + + sum := 0 + for _, v := range got { + sum += v + } + if sum != 60 { + t.Fatalf("expected sum 60, got %d", sum) + } +} + +func TestRun_Backpressure(t *testing.T) { + t.Parallel() + + input := make(chan int) + consumed := make(chan int, 10) + + out := Run(t.Context(), input, + Map(1, func(_ context.Context, v int) (int, bool) { + consumed <- v + return v, true + }), + ) + + input <- 42 + v := <-consumed + if v != 42 { + t.Fatalf("expected 42, got %d", v) + } + + select { + case input <- 99: + t.Fatal("producer was not blocked") + case <-time.After(50 * time.Millisecond): + } + + go func() { + for range out { + } + }() + close(input) +} + +func TestRun_GracefulDrain(t *testing.T) { + t.Parallel() + + input := make(chan int, 5) + for i := 1; i <= 5; i++ { + input <- i + } + close(input) + + var count atomic.Int32 + out := Run(t.Context(), input, + Sink(2, func(_ context.Context, v int) { + count.Add(1) + }), + ) + + for range out { + } + + if count.Load() != 5 { + t.Fatalf("expected all 5 items drained, got %d", count.Load()) + } +} + +func TestRun_Cancellation(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + + input := make(chan int) + out := Run(ctx, input, + Map(1, func(_ context.Context, v int) (int, bool) { + return v, true + }), + ) + + go func() { + select { + case input <- 1: + case <-ctx.Done(): + } + }() + + cancel() + + done := make(chan struct{}) + go func() { + for range out { + } + close(done) + }() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("pipeline did not shut down after cancellation") + } +} + +func TestRun_ErrorPropagation(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancelCause(t.Context()) + defer cancel(nil) + + input := make(chan int, 5) + for i := 1; i <= 5; i++ { + input <- i + } + close(input) + + sentinel := fmt.Errorf("bad item") + + out := Run(ctx, input, + Map(1, func(_ context.Context, v int) (int, bool) { + if v == 3 { + cancel(sentinel) + return v, false + } + return v * 2, true + }), + ) + + for range out { + } + + if cause := context.Cause(ctx); cause != sentinel { + t.Fatalf("expected sentinel error, got: %v", cause) + } +} + +func TestObserver(t *testing.T) { + t.Parallel() + + var items, drops atomic.Int32 + doneCh := make(chan struct{}) + + obs := &testObserver{ + onItem: func() { items.Add(1) }, + onDrop: func() { drops.Add(1) }, + onDone: func() { close(doneCh) }, + } + + input := make(chan int, 6) + for i := 1; i <= 6; i++ { + input <- i + } + close(input) + + out := Run(t.Context(), input, + Filter(func(v int) bool { return v%2 == 0 }, WithObserver(obs)), + ) + + for range out { + } + <-doneCh + + if items.Load() != 3 { + t.Errorf("expected 3 items, got %d", items.Load()) + } + if drops.Load() != 3 { + t.Errorf("expected 3 drops, got %d", drops.Load()) + } +} + +type testObserver struct { + onItem func() + onDrop func() + onDone func() +} + +func (o *testObserver) OnItem() { + if o.onItem != nil { + o.onItem() + } +} + +func (o *testObserver) OnDrop() { + if o.onDrop != nil { + o.onDrop() + } +} + +func (o *testObserver) OnDone() { + if o.onDone != nil { + o.onDone() + } +} diff --git a/pipeline/stage.go b/pipeline/stage.go new file mode 100644 index 0000000..e0716cf --- /dev/null +++ b/pipeline/stage.go @@ -0,0 +1,310 @@ +package pipeline + +import ( + "context" + "sync" +) + +// Map applies fn to each item with bounded concurrency. +// If keep is false, the item is dropped. +func Map[T any](concurrency int, fn func(context.Context, T) (T, bool), opts ...StageOption) Stage[T] { + if concurrency < 1 { + panic("pipeline: Map concurrency must be at least 1") + } + cfg := applyOpts(opts) + + return func(ctx context.Context, in <-chan T) <-chan T { + out := make(chan T, cfg.bufferSize) + var wg sync.WaitGroup + + for range concurrency { + wg.Go(func() { + for { + select { + case v, ok := <-in: + if !ok { + return + } + result, keep := fn(ctx, v) + if !keep { + if cfg.observer != nil { + cfg.observer.OnDrop() + } + continue + } + if cfg.observer != nil { + cfg.observer.OnItem() + } + select { + case out <- result: + case <-ctx.Done(): + return + } + case <-ctx.Done(): + return + } + } + }) + } + + go func() { + wg.Wait() + close(out) + if cfg.observer != nil { + cfg.observer.OnDone() + } + }() + + return out + } +} + +// Filter forwards items for which fn returns true. +func Filter[T any](fn func(T) bool, opts ...StageOption) Stage[T] { + cfg := applyOpts(opts) + + return func(ctx context.Context, in <-chan T) <-chan T { + out := make(chan T, cfg.bufferSize) + + go func() { + defer close(out) + defer func() { + if cfg.observer != nil { + cfg.observer.OnDone() + } + }() + + for { + select { + case v, ok := <-in: + if !ok { + return + } + if !fn(v) { + if cfg.observer != nil { + cfg.observer.OnDrop() + } + continue + } + if cfg.observer != nil { + cfg.observer.OnItem() + } + select { + case out <- v: + case <-ctx.Done(): + return + } + case <-ctx.Done(): + return + } + } + }() + + return out + } +} + +// ForEach calls fn on each item and forwards it unchanged. +func ForEach[T any](fn func(T), opts ...StageOption) Stage[T] { + cfg := applyOpts(opts) + + return func(ctx context.Context, in <-chan T) <-chan T { + out := make(chan T, cfg.bufferSize) + + go func() { + defer close(out) + defer func() { + if cfg.observer != nil { + cfg.observer.OnDone() + } + }() + + for { + select { + case v, ok := <-in: + if !ok { + return + } + fn(v) + if cfg.observer != nil { + cfg.observer.OnItem() + } + select { + case out <- v: + case <-ctx.Done(): + return + } + case <-ctx.Done(): + return + } + } + }() + + return out + } +} + +// Buffer inserts a buffered channel of the specified capacity between stages. +func Buffer[T any](size int) Stage[T] { + if size < 1 { + panic("pipeline: Buffer size must be at least 1") + } + + return func(ctx context.Context, in <-chan T) <-chan T { + out := make(chan T, size) + + go func() { + defer close(out) + for { + select { + case v, ok := <-in: + if !ok { + return + } + select { + case out <- v: + case <-ctx.Done(): + return + } + case <-ctx.Done(): + return + } + } + }() + + return out + } +} + +// Tee sends each item to a side channel while forwarding it to the next stage. +func Tee[T any](side chan<- T, opts ...StageOption) Stage[T] { + if side == nil { + panic("pipeline: Tee side channel must not be nil") + } + cfg := applyOpts(opts) + + return func(ctx context.Context, in <-chan T) <-chan T { + out := make(chan T, cfg.bufferSize) + + go func() { + defer close(out) + defer func() { + if cfg.observer != nil { + cfg.observer.OnDone() + } + }() + + for { + select { + case v, ok := <-in: + if !ok { + return + } + if cfg.observer != nil { + cfg.observer.OnItem() + } + select { + case side <- v: + case <-ctx.Done(): + return + } + select { + case out <- v: + case <-ctx.Done(): + return + } + case <-ctx.Done(): + return + } + } + }() + + return out + } +} + +// Take forwards up to n items, then closes the output channel. +func Take[T any](n int, opts ...StageOption) Stage[T] { + if n < 0 { + panic("pipeline: Take count must not be negative") + } + cfg := applyOpts(opts) + + return func(ctx context.Context, in <-chan T) <-chan T { + out := make(chan T, cfg.bufferSize) + + go func() { + defer close(out) + defer func() { + if cfg.observer != nil { + cfg.observer.OnDone() + } + }() + + count := 0 + for count < n { + select { + case v, ok := <-in: + if !ok { + return + } + if cfg.observer != nil { + cfg.observer.OnItem() + } + select { + case out <- v: + count++ + case <-ctx.Done(): + return + } + case <-ctx.Done(): + return + } + } + }() + + return out + } +} + +// Sink consumes items with bounded concurrency without forwarding them. +func Sink[T any](concurrency int, fn func(context.Context, T), opts ...StageOption) Stage[T] { + if concurrency < 1 { + panic("pipeline: Sink concurrency must be at least 1") + } + cfg := applyOpts(opts) + + return func(ctx context.Context, in <-chan T) <-chan T { + done := make(chan T) + var wg sync.WaitGroup + + for range concurrency { + wg.Go(func() { + for { + select { + case v, ok := <-in: + if !ok { + return + } + fn(ctx, v) + if cfg.observer != nil { + cfg.observer.OnItem() + } + case <-ctx.Done(): + return + } + } + }) + } + + go func() { + wg.Wait() + close(done) + if cfg.observer != nil { + cfg.observer.OnDone() + } + }() + + return done + } +} diff --git a/pipeline/stage_test.go b/pipeline/stage_test.go new file mode 100644 index 0000000..4579158 --- /dev/null +++ b/pipeline/stage_test.go @@ -0,0 +1,308 @@ +package pipeline + +import ( + "context" + "sort" + "sync" + "sync/atomic" + "testing" +) + +func TestMap_Basic(t *testing.T) { + t.Parallel() + + input := make(chan int, 3) + input <- 1 + input <- 2 + input <- 3 + close(input) + + out := Map(1, func(_ context.Context, v int) (int, bool) { + return v * 10, true + })(t.Context(), input) + + var got []int + for v := range out { + got = append(got, v) + } + + sort.Ints(got) + if len(got) != 3 || got[0] != 10 || got[1] != 20 || got[2] != 30 { + t.Fatalf("expected [10 20 30], got %v", got) + } +} + +func TestMap_Concurrent(t *testing.T) { + t.Parallel() + + input := make(chan int, 100) + for i := range 100 { + input <- i + } + close(input) + + out := Map(5, func(_ context.Context, v int) (int, bool) { + return v, true + })(t.Context(), input) + + var got []int + for v := range out { + got = append(got, v) + } + + if len(got) != 100 { + t.Fatalf("expected 100 items, got %d", len(got)) + } + + sort.Ints(got) + for i, v := range got { + if v != i { + t.Fatalf("missing item %d, got %v at position %d", i, v, i) + } + } +} + +func TestMap_Drop(t *testing.T) { + t.Parallel() + + input := make(chan int, 5) + for i := 1; i <= 5; i++ { + input <- i + } + close(input) + + out := Map(1, func(_ context.Context, v int) (int, bool) { + return v, v > 3 + })(t.Context(), input) + + var got []int + for v := range out { + got = append(got, v) + } + + if len(got) != 2 { + t.Fatalf("expected 2 items, got %d: %v", len(got), got) + } +} + +func TestMap_PanicOnZeroConcurrency(t *testing.T) { + t.Parallel() + + defer func() { + if r := recover(); r == nil { + t.Error("expected panic for concurrency < 1") + } + }() + + Map(0, func(_ context.Context, v int) (int, bool) { return v, true }) +} + +func TestFilter(t *testing.T) { + t.Parallel() + + input := make(chan int, 10) + for i := 1; i <= 10; i++ { + input <- i + } + close(input) + + out := Filter(func(v int) bool { return v%3 == 0 })(t.Context(), input) + + var got []int + for v := range out { + got = append(got, v) + } + + if len(got) != 3 { + t.Fatalf("expected [3 6 9], got %v", got) + } +} + +func TestForEach(t *testing.T) { + t.Parallel() + + input := make(chan int, 3) + input <- 1 + input <- 2 + input <- 3 + close(input) + + var sum atomic.Int32 + out := ForEach(func(v int) { + sum.Add(int32(v)) + })(t.Context(), input) + + var got []int + for v := range out { + got = append(got, v) + } + + if sum.Load() != 6 { + t.Fatalf("expected side effect sum 6, got %d", sum.Load()) + } + if len(got) != 3 { + t.Fatalf("expected 3 forwarded items, got %d", len(got)) + } +} + +func TestBuffer(t *testing.T) { + t.Parallel() + + input := make(chan int, 5) + for i := 1; i <= 5; i++ { + input <- i + } + close(input) + + out := Buffer[int](10)(t.Context(), input) + + var got []int + for v := range out { + got = append(got, v) + } + + if len(got) != 5 { + t.Fatalf("expected 5 items, got %d", len(got)) + } +} + +func TestBuffer_PanicOnZero(t *testing.T) { + t.Parallel() + + defer func() { + if r := recover(); r == nil { + t.Error("expected panic for buffer size < 1") + } + }() + + Buffer[int](0) +} + +func TestTee(t *testing.T) { + t.Parallel() + + input := make(chan int, 3) + input <- 1 + input <- 2 + input <- 3 + close(input) + + side := make(chan int, 10) + out := Tee(side)(t.Context(), input) + + var got []int + for v := range out { + got = append(got, v) + } + close(side) + + var sideGot []int + for v := range side { + sideGot = append(sideGot, v) + } + + if len(got) != 3 { + t.Fatalf("expected 3 forwarded items, got %d", len(got)) + } + if len(sideGot) != 3 { + t.Fatalf("expected 3 side items, got %d", len(sideGot)) + } +} + +func TestTee_NilSide(t *testing.T) { + t.Parallel() + + defer func() { + if r := recover(); r == nil { + t.Error("expected panic for nil side channel") + } + }() + + Tee[int](nil) +} + +func TestTake(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + input := make(chan int, 10) + for i := 1; i <= 10; i++ { + input <- i + } + close(input) + + out := Take[int](3)(ctx, input) + + var got []int + for v := range out { + got = append(got, v) + } + + if len(got) != 3 { + t.Fatalf("expected 3 items, got %d: %v", len(got), got) + } +} + +func TestTake_Zero(t *testing.T) { + t.Parallel() + + input := make(chan int, 5) + for i := 1; i <= 5; i++ { + input <- i + } + close(input) + + out := Take[int](0)(t.Context(), input) + + count := 0 + for range out { + count++ + } + + if count != 0 { + t.Fatalf("expected 0 items from Take(0), got %d", count) + } +} + +func TestSink(t *testing.T) { + t.Parallel() + + input := make(chan int, 5) + for i := 1; i <= 5; i++ { + input <- i + } + close(input) + + var mu sync.Mutex + var collected []int + + out := Sink(2, func(_ context.Context, v int) { + mu.Lock() + collected = append(collected, v) + mu.Unlock() + })(t.Context(), input) + + for range out { + t.Fatal("Sink should not forward items") + } + + mu.Lock() + defer mu.Unlock() + + if len(collected) != 5 { + t.Fatalf("expected 5 items consumed, got %d", len(collected)) + } +} + +func TestSink_PanicOnZeroConcurrency(t *testing.T) { + t.Parallel() + + defer func() { + if r := recover(); r == nil { + t.Error("expected panic for concurrency < 1") + } + }() + + Sink(0, func(_ context.Context, v int) {}) +} -- cgit v1.3.1