go fix, and modernize some code

broadcast.go

@@ -2,6 +2,7 @@ package events
 
 import (
 	"fmt"
+	"slices"
 	"sync"
 
 	"github.com/sirupsen/logrus"
@@ -105,55 +106,46 @@ func (b *Broadcaster) Close() error {
 // Close is called, this goroutine will exit.
 func (b *Broadcaster) run() {
 	defer close(b.closed)
-	remove := func(target Sink) {
-		for i, sink := range b.sinks {
-			if sink == target {
-				b.sinks = append(b.sinks[:i], b.sinks[i+1:]...)
-				break
-			}
-		}
-	}
 
 	for {
 		select {
 		case event := <-b.events:
-			for _, sink := range b.sinks {
+			for i := 0; i < len(b.sinks); {
+				sink := b.sinks[i]
 				if err := sink.Write(event); err != nil {
 					if err == ErrSinkClosed {
-						// remove closed sinks
-						remove(sink)
+						b.sinks = slices.Delete(b.sinks, i, i+1)
 						continue
 					}
-					logrus.WithField("event", event).WithField("events.sink", sink).WithError(err).
-						Errorf("broadcaster: dropping event")
+					logrus.WithFields(logrus.Fields{
+						"error":       err,
+						"event":       event,
+						"events.sink": sink,
+					}).Error("broadcaster: dropping event")
 				}
+				i++
 			}
 		case request := <-b.adds:
 			// while we have to iterate for add/remove, common iteration for
 			// send is faster against slice.
-
-			var found bool
-			for _, sink := range b.sinks {
-				if request.sink == sink {
-					found = true
-					break
-				}
-			}
-
-			if !found {
+			// b.sinks[request.sink] = struct{}{}
+			if !slices.Contains(b.sinks, request.sink) {
 				b.sinks = append(b.sinks, request.sink)
 			}
-			// b.sinks[request.sink] = struct{}{}
 			request.response <- nil
 		case request := <-b.removes:
-			remove(request.sink)
+			if i := slices.Index(b.sinks, request.sink); i >= 0 {
+				b.sinks = slices.Delete(b.sinks, i, i+1)
+			}
 			request.response <- nil
 		case <-b.shutdown:
 			// close all the underlying sinks
 			for _, sink := range b.sinks {
 				if err := sink.Close(); err != nil && err != ErrSinkClosed {
-					logrus.WithField("events.sink", sink).WithError(err).
-						Errorf("broadcaster: closing sink failed")
+					logrus.WithFields(logrus.Fields{
+						"error":       err,
+						"events.sink": sink,
+					}).Error("broadcaster: closing sink failed")
 				}
 			}
 			return
@@ -164,16 +156,13 @@ func (b *Broadcaster) run() {
 func (b *Broadcaster) String() string {
 	// Serialize copy of this broadcaster without the sync.Once, to avoid
 	// a data race.
-
-	b2 := map[string]interface{}{
+	return fmt.Sprint(map[string]any{
 		"sinks":   b.sinks,
 		"events":  b.events,
 		"adds":    b.adds,
 		"removes": b.removes,
 
 		"shutdown": b.shutdown,
 		"closed":   b.closed,
-	}
-
-	return fmt.Sprint(b2)
+	})
 }

broadcast_test.go

@@ -10,7 +10,7 @@ func TestBroadcaster(t *testing.T) {
 	const nEvents = 1000
 	var sinks []Sink
 	b := NewBroadcaster()
-	for i := 0; i < 10; i++ {
+	for i := range 10 {
 		sinks = append(sinks, newTestSink(t, nEvents))
 		err := b.Add(sinks[i])
 		if err != nil {
@@ -108,7 +108,7 @@ func benchmarkBroadcast(b *testing.B, nsinks int) {
 
 	b.StopTimer()
 	var sinks []Sink
-	for i := 0; i < nsinks; i++ {
+	for range nsinks {
 		// counter.Inc(1)
 		sinks = append(sinks, newTestSink(b, b.N))
 		// sinks = append(sinks, NewQueue(&testSink{t: b, expected: b.N}))

channel.go

@@ -53,9 +53,8 @@ func (ch *Channel) Close() error {
 func (ch *Channel) String() string {
 	// Serialize a copy of the Channel that doesn't contain the sync.Once,
 	// to avoid a data race.
-	ch2 := map[string]interface{}{
+	return fmt.Sprint(map[string]any{
 		"C":      ch.C,
 		"closed": ch.closed,
-	}
-	return fmt.Sprint(ch2)
+	})
 }

common_test.go

@@ -2,15 +2,15 @@ package events
 
 import (
 	"fmt"
-	"math/rand"
+	"math/rand/v2"
 	"sync"
 	"testing"
 	"time"
 )
 
 type tOrB interface {
-	Fatalf(format string, args ...interface{})
-	Logf(format string, args ...interface{})
+	Fatalf(format string, args ...any)
+	Logf(format string, args ...any)
 }
 
 type testSink struct {

event.go

@@ -1,7 +1,7 @@
 package events
 
 // Event marks items that can be sent as events.
-type Event interface{}
+type Event any
 
 // Sink accepts and sends events.
 type Sink interface {

filter_test.go

@@ -10,7 +10,7 @@ func TestFilter(t *testing.T) {
 		return ok && i%2 == 0
 	}))
 
-	for i := 0; i < nevents; i++ {
+	for i := range nevents {
 		if err := filter.Write(i); err != nil {
 			t.Fatalf("unexpected error writing event: %v", err)
 		}

queue.go

@@ -103,6 +103,7 @@ func (eq *Queue) next() Event {
 		eq.cond.Wait()
 	}
 
+	// Len is non-zero while holding eq.mu, so Front cannot be nil.
 	front := eq.events.Front()
 	block := front.Value.(Event)
 	eq.events.Remove(front)

retry.go

@@ -2,7 +2,7 @@ package events
 
 import (
 	"fmt"
-	"math/rand"
+	"math/rand/v2"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -26,13 +26,11 @@ type RetryingSink struct {
 // off on failure. Parameters threshold and backoff adjust the behavior of the
 // circuit breaker.
 func NewRetryingSink(sink Sink, strategy RetryStrategy) *RetryingSink {
-	rs := &RetryingSink{
+	return &RetryingSink{
 		sink:     sink,
 		strategy: strategy,
 		closed:   make(chan struct{}),
 	}
-
-	return rs
 }
 
 // Write attempts to flush the events to the downstream sink until it succeeds
@@ -66,14 +64,12 @@ retry:
 			return err
 		}
 
-		logger := logger.WithError(err) // shadow!!
-
 		if rs.strategy.Failure(event, err) {
-			logger.Errorf("retryingsink: dropped event")
+			logger.WithError(err).Error("retryingsink: dropped event")
 			return nil
 		}
 
-		logger.Errorf("retryingsink: error writing event, retrying")
+		logger.WithError(err).Error("retryingsink: error writing event, retrying")
 		goto retry
 	}
 
@@ -93,7 +89,7 @@ func (rs *RetryingSink) Close() error {
 func (rs *RetryingSink) String() string {
 	// Serialize a copy of the RetryingSink without the sync.Once, to avoid
 	// a data race.
-	rs2 := map[string]interface{}{
+	rs2 := map[string]any{
 		"sink":     rs.sink,
 		"strategy": rs.strategy,
 		"closed":   rs.closed,
@@ -201,7 +197,7 @@ type ExponentialBackoffConfig struct {
 // ExponentialBackoff implements random backoff with exponentially increasing
 // bounds as the number consecutive failures increase.
 type ExponentialBackoff struct {
-	failures uint64 // consecutive failure counter (needs to be 64-bit aligned)
+	failures atomic.Uint64 // consecutive failure counter (needs to be 64-bit aligned)
 	config   ExponentialBackoffConfig
 }
 
@@ -215,17 +211,17 @@ func NewExponentialBackoff(config ExponentialBackoffConfig) *ExponentialBackoff
 
 // Proceed returns the next randomly bound exponential backoff time.
 func (b *ExponentialBackoff) Proceed(event Event) time.Duration {
-	return b.backoff(atomic.LoadUint64(&b.failures))
+	return b.backoff(b.failures.Load())
 }
 
 // Success resets the failures counter.
 func (b *ExponentialBackoff) Success(event Event) {
-	atomic.StoreUint64(&b.failures, 0)
+	b.failures.Store(0)
 }
 
 // Failure increments the failure counter.
 func (b *ExponentialBackoff) Failure(event Event, err error) bool {
-	atomic.AddUint64(&b.failures, 1)
+	b.failures.Add(1)
 	return false
 }
 
@@ -242,17 +238,16 @@ func (b *ExponentialBackoff) backoff(failures uint64) time.Duration {
 		factor = DefaultExponentialBackoffConfig.Factor
 	}
 
-	backoff := b.config.Base + factor*time.Duration(1<<(failures-1))
-
-	max := b.config.Max
-	if max <= 0 {
-		max = DefaultExponentialBackoffConfig.Max
+	maxBackoff := b.config.Max
+	if maxBackoff <= 0 {
+		maxBackoff = DefaultExponentialBackoffConfig.Max
 	}
 
-	if backoff > max || backoff < 0 {
-		backoff = max
+	backoff := b.config.Base + factor*time.Duration(1<<(failures-1))
+	if backoff > maxBackoff || backoff < 0 {
+		backoff = maxBackoff
 	}
 
 	// Choose a uniformly distributed value from [0, backoff).
-	return time.Duration(rand.Int63n(int64(backoff)))
+	return rand.N(backoff)
 }

retry_test.go

@@ -84,18 +84,15 @@ func TestExponentialBackoff(t *testing.T) {
 			t.Errorf("no facilities for dropping events in ExponentialBackoff")
 		}
 
-		for j := 0; j < 1000; j++ {
+		for range 1000 {
 			// sample this several thousand times.
 			backoff := strategy.Proceed(nil)
 			if backoff > expected {
 				t.Fatalf("expected must be bounded by %v after %v failures: %v", expected, i, backoff)
 			}
 		}
 
-		expected = strategy.config.Base + strategy.config.Factor*time.Duration(1<<uint64(i))
-		if expected > strategy.config.Max {
-			expected = strategy.config.Max
-		}
+		expected = min(strategy.config.Base+strategy.config.Factor*time.Duration(1<<uint64(i)), strategy.config.Max)
 	}
 
 	strategy.Success(nil) // recovery!