From 03878d151215277d948b0a1ce31c65d9305d3aa4 Mon Sep 17 00:00:00 2001
From: DJ O'Leary <dijitol@proton.me>
Date: Tue, 14 Jul 2026 02:48:22 +0200
Subject: refactor(pkg): move packages that can be imported by others to pkg
 directory

---
 pkg/sync/chan_pipeline.go | 48 +++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 48 insertions(+)
 create mode 100644 pkg/sync/chan_pipeline.go

(limited to 'pkg/sync/chan_pipeline.go')

diff --git a/pkg/sync/chan_pipeline.go b/pkg/sync/chan_pipeline.go
new file mode 100644
index 0000000..6b5f37f
--- /dev/null
+++ b/pkg/sync/chan_pipeline.go
@@ -0,0 +1,48 @@
+package sync
+
+import (
+	"context"
+	"sync"
+)
+
+func Pipe[IN, OUT any](
+	ctx context.Context,
+	in <-chan IN,
+	transform func(context.Context, IN) OUT,
+	workerCount int,
+	buffer int,
+) <-chan OUT {
+	if in == nil {
+		// we only expect this to happen during development
+		panic("in-channel is nil")
+	}
+	if workerCount < 1 {
+		workerCount = 1
+	}
+	if buffer < 0 {
+		buffer = 0
+	}
+
+	wg := sync.WaitGroup{}
+
+	out := make(chan OUT, buffer)
+
+	for range workerCount {
+		wg.Go(func() {
+			for v := range OrDone(ctx, in) {
+				w := transform(ctx, v)
+
+				if !CancelOrSend(ctx, out, w) {
+					return
+				}
+			}
+		})
+	}
+
+	go func() {
+		wg.Wait()
+		close(out)
+	}()
+
+	return out
+}
-- 
cgit v1.2.3