From 02123df8cc15896f97c1580e49d8e2a991e37ddd Mon Sep 17 00:00:00 2001
From: sangtran <tranquangsang12.7@gmail.com>
Date: Thu, 3 Sep 2026 23:10:33 +0700
Subject: Moving all legacy quacker source code from github to cgit

---
 pipeline/batch.go | 64 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 64 insertions(+)
 create mode 100644 pipeline/batch.go

(limited to 'pipeline/batch.go')

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
+}
-- 
cgit v1.3.1