1 files changed, 48 insertions, 0 deletions
diff --git a/pkg/sync/chan.go b/pkg/sync/chan.go
new file mode 100644
index 0000000..91501f9
--- /dev/null
+++ b/pkg/sync/chan.go
@@ -0,0 +1,48 @@
+package sync
+
+import "context"
+
+func OrDone[T any](ctx context.Context, in <-chan T) <-chan T {
+	if in == nil {
+		// we only expect this to happen during development
+		panic("in-channel is nil")
+	}
+
+	out := make(chan T)
+	go func() {
+		defer close(out)
+		for {
+			v, ok := CancelOrReceive(ctx, in)
+			if !ok {
+				return
+			}
+
+			if !CancelOrSend(ctx, out, v) {
+				return
+			}
+		}
+	}()
+	return out
+}
+
+// CancelOrReceive is trying to solve the same problem as OrDone but does so
+// without an extra goroutine.
+func CancelOrReceive[T any](ctx context.Context, in <-chan T) (v T, ok bool) {
+	select {
+	case <-ctx.Done():
+		return // <zero value of type T>, false
+	case v, ok = <-in:
+		return // v, ok
+	}
+}
+
+// CancelOrSend is a helper function to simplify sending values to channels
+// while keeping the context in mind.
+func CancelOrSend[T any](ctx context.Context, out chan<- T, val T) bool {
+	select {
+	case <-ctx.Done():
+		return false
+	case out <- val:
+		return true
+	}
+}