aboutsummaryrefslogtreecommitdiff
path: root/channels/fanin.go
diff options
context:
space:
mode:
Diffstat (limited to 'channels/fanin.go')
-rw-r--r--channels/fanin.go23
1 files changed, 23 insertions, 0 deletions
diff --git a/channels/fanin.go b/channels/fanin.go
new file mode 100644
index 0000000..679aeee
--- /dev/null
+++ b/channels/fanin.go
@@ -0,0 +1,23 @@
+package channels
+
+import "sync/atomic"
+
+// FanIn сливает несколько каналов в один
+func FanIn[T any](chans ...chan T) chan T {
+ out := make(chan T)
+ workers := int64(len(chans))
+ for _, ch := range chans {
+ func(ch chan T) {
+ go func() {
+ for t := range ch {
+ out <- t
+ }
+ atomic.AddInt64(&workers, -1)
+ if workers == 0 {
+ close(out)
+ }
+ }()
+ }(ch)
+ }
+ return out
+}