implemented the SyncTopic

async_test.go

@@ -35,14 +35,14 @@ func TestAsyncTopic_SinglePublisherSingleSubscriber(t *testing.T) {
 	}
 
 	count := 0
-	timeout := time.After(time.Second)
+	timeout := testTimer(t, time.Second)
 
 	for count < msgCount {
 		select {
 		case <-feedback:
 			count++
 
-		case <-timeout:
+		case <-timeout.C:
 			t.Fatalf("expected %d feedback items by now but only got %d", msgCount, count)
 		}
 	}
@@ -93,14 +93,14 @@ func TestAsyncTopic_MultiPublishersMultiSubscribers(t *testing.T) {
 	}
 
 	count := 0
-	timeout := time.After(time.Second)
+	timeout := testTimer(t, time.Second)
 
 	for count != expFeedbackCount {
 		select {
 		case <-feedback:
 			count++
 
-		case <-timeout:
+		case <-timeout.C:
 			t.Fatalf("expected %d feedback items by now but only got %d", expFeedbackCount, count)
 		}
 	}
@@ -121,7 +121,7 @@ func TestAsyncTopic_WithOnClose(t *testing.T) {
 	case <-feedback:
 		break
 
-	case <-time.After(time.Second):
+	case <-testTimer(t, time.Second).C:
 		t.Fatalf("expected feedback by now")
 	}
 }
@@ -142,7 +142,7 @@ func TestAsyncTopic_WithOnSubscribe(t *testing.T) {
 	}
 
 	count := 0
-	timeout := time.After(time.Second)
+	timeout := testTimer(t, time.Second)
 
 	for count < totalSub {
 		select {
@@ -150,7 +150,7 @@ func TestAsyncTopic_WithOnSubscribe(t *testing.T) {
 			count++
 			assert.Equal(t, count, c, "expected %d but got %d instead", count, c)
 
-		case <-timeout:
+		case <-timeout.C:
 			t.Fatalf("expected %d feedback items by now but only got %d", totalSub, count)
 		}
 	}
@@ -190,7 +190,7 @@ func TestAsyncTopic_ClosedTopicError(t *testing.T) {
 			case <-feedback:
 				break
 
-			case <-time.After(time.Second):
+			case <-testTimer(t, time.Second).C:
 				t.Fatalf("expected feedback by now")
 			}
 
@@ -230,15 +230,26 @@ func TestAsyncTopic_AllPublishedBeforeClosedAreDeliveredAfterClosed(t *testing.T
 	cancel()
 
 	values := make(map[int]struct{}, msgCount)
-	timeout := time.After(time.Second)
+	timeout := testTimer(t, time.Second)
 
 	for len(values) < msgCount {
 		select {
 		case f := <-feedback:
 			values[f] = struct{}{}
 
-		case <-timeout:
+		case <-timeout.C:
 			t.Fatalf("expected %d unique feedback values by now but only got %d", msgCount, len(values))
 		}
 	}
 }
+
+func testTimer(t testing.TB, d time.Duration) *time.Timer {
+	t.Helper()
+
+	timer := time.NewTimer(d)
+	t.Cleanup(func() {
+		timer.Stop()
+	})
+
+	return timer
+}

sync.go

@@ -0,0 +1,44 @@
+package gubgub
+
+import "sync"
+
+// SyncTopic allows any message T to be broadcast to subscribers. Publishing and Subscribing
+// happens synchronously (block).
+type SyncTopic[T any] struct {
+	mu          sync.Mutex
+	subscribers []Subscriber[T]
+}
+
+// NewSyncTopic creates a zero SyncTopic and return a pointer to it.
+func NewSyncTopic[T any]() *SyncTopic[T] {
+	return &SyncTopic[T]{}
+}
+
+// Publish broadcasts a message to all subscribers.
+func (t *SyncTopic[T]) Publish(msg T) error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	keepers := make([]Subscriber[T], 0, len(t.subscribers))
+
+	for _, callback := range t.subscribers {
+		keep := callback(msg)
+		if keep {
+			keepers = append(keepers, callback)
+		}
+	}
+
+	t.subscribers = keepers
+
+	return nil
+}
+
+// Subscribe adds a Subscriber func that will consume future published messages.
+func (t *SyncTopic[T]) Subscribe(fn Subscriber[T]) error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	t.subscribers = append(t.subscribers, fn)
+
+	return nil
+}

sync_test.go

@@ -0,0 +1,71 @@
+package gubgub
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSyncTopic_SinglePublisherSingleSubscriber(t *testing.T) {
+	const msgCount = 10
+
+	topic := NewSyncTopic[int]()
+
+	var feedback []int
+
+	err := topic.Subscribe(func(i int) bool {
+		feedback = append(feedback, i)
+		return true
+	})
+	require.NoError(t, err)
+
+	for i := range msgCount {
+		require.NoError(t, topic.Publish(i))
+	}
+
+	assert.Len(t, feedback, msgCount, "missing some feedback values")
+}
+
+func TestSyncTopic_MultiPublishersMultiSubscribers(t *testing.T) {
+	const (
+		subCount = 10
+		pubCount = 10
+		msgCount = pubCount * 100 // total messages to publish (delivered to EACH subscriber)
+	)
+
+	topic := NewSyncTopic[int]()
+
+	var feedbackCounter int
+
+	for range subCount {
+		err := topic.Subscribe(func(i int) bool {
+			feedbackCounter++
+			return true
+		})
+		require.NoError(t, err)
+	}
+
+	toDeliver := make(chan int, msgCount)
+	for i := range msgCount {
+		toDeliver <- i
+	}
+	close(toDeliver)
+
+	var wg sync.WaitGroup
+
+	wg.Add(pubCount)
+	for range pubCount {
+		go func() {
+			defer wg.Done()
+			for msg := range toDeliver {
+				require.NoError(t, topic.Publish(msg))
+			}
+		}()
+	}
+
+	wg.Wait()
+
+	assert.Equal(t, msgCount*subCount, feedbackCounter)
+}