add Broadcaster

Author: smallnest

README.md

@@ -25,6 +25,7 @@ Experimental packages not in std and golang.org/exp
     - Uint32
     - Uint64
   - **Pool**: a generic pool, forked from https://github.com/mkmik/syncpool
+  - **Broadcaster**: a broadcaster to broadcast message to multiple receivers
 
 - **container**
   - **heap**: generic heap

sync/broadcaster.go

@@ -0,0 +1,46 @@
+package sync
+
+import "sync"
+
+type Broadcaster[T any] struct {
+	mu       *sync.Mutex
+	cond     *sync.Cond
+	signaled bool
+
+	v T
+}
+
+// Broadcast sends a signal with a value to all goroutines waiting on the broadcaster.
+func NewBroadcaster[T any]() *Broadcaster[T] {
+	var mu sync.Mutex
+	return &Broadcaster[T]{
+		mu:       &mu,
+		cond:     sync.NewCond(&mu),
+		signaled: false,
+	}
+}
+
+// Go waits until something is broadcasted, and runs the given function in a new
+// goroutine with the value that was broadcasted.
+func (b *Broadcaster[T]) Go(fn func(v T)) {
+	go func() {
+		b.cond.L.Lock()
+		defer b.cond.L.Unlock()
+
+		for !b.signaled {
+			b.cond.Wait()
+		}
+		fn(b.v)
+	}()
+}
+
+// Broadcast broadcasts a signal to all
+// waiting function and unblocks them.
+func (b *Broadcaster[T]) Broadcast(v T) {
+	b.cond.L.Lock()
+	b.v = v
+	b.signaled = true
+	b.cond.L.Unlock()
+
+	b.cond.Broadcast()
+}

sync/broadcaster_test.go

@@ -0,0 +1,36 @@
+package sync
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewBroadcaster(t *testing.T) {
+	b := NewBroadcaster[int]()
+
+	var count atomic.Int32
+	var sum atomic.Int64
+
+	var wg sync.WaitGroup
+
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		b.Go(func(v int) {
+			defer wg.Done()
+
+			count.Add(1)
+			sum.Add(int64(v))
+		})
+	}
+
+	b.Broadcast(10)
+
+	wg.Wait()
+
+	assert.Equal(t, int32(10), count.Load())
+	assert.Equal(t, int64(100), sum.Load())
+
+}