Release 1.0.2

Author: badu

README.md

@@ -24,6 +24,8 @@ or
 
 By default, the bus is using sync events : waits for listeners to complete their jobs before calling the next listener.
 
+Or, even easier, just call `PubAsync` methods.
+
 Usage : `go get github.com/badu/bus`
 
 ## What Problem Does It Solve?
@@ -53,9 +55,9 @@ Inside the `test_scenarios` folder, you can find the following scenarios:
    SMS) and `audit`. When a user registers, we want to send welcoming messages via SMS and email, but we also want to
    audit that registration for reporting purposes.
 
-   The [UserRegisteredEvent](https://github.com/badu/bus/blob/main/test_scenarios/fire-and-forget/events/main.go#L10)
+   The [UserRegisteredEvent](https://github.com/badu/bus/blob/main/test_scenarios/fire-and-forget/events/main.go#L3)
    will carry the freshly registered username (which is also the email) and phone to the email and sms services. The
-   event is [triggered](https://github.com/badu/bus/blob/main/test_scenarios/fire-and-forget/users/service.go#L21) by
+   event is [triggered](https://github.com/badu/bus/blob/main/test_scenarios/fire-and-forget/users/service.go#L23) by
    the user service, which performs the creation of the user account. We're using the `fire and forget` technique here,
    because the operation of registration should not depend on the fact that we've been able to
    send an welcoming email or a sms, or the audit system malfunctions.

bench_test.go

@@ -11,18 +11,10 @@ type Uint32SyncEvent struct {
 	u uint32
 }
 
-func (u Uint32SyncEvent) EventID() string {
-	return "Uint32SyncEvent"
-}
-
 type Uint32AsyncEvent struct {
 	u uint32
 }
 
-func (u Uint32AsyncEvent) EventID() string {
-	return "Uint32AsyncEvent"
-}
-
 func BenchmarkBroadcast_0008Sync(b *testing.B) {
 	topic := bus.NewTopic[Uint32SyncEvent]()
 	c := uint32(0)

main.go

@@ -13,6 +13,11 @@ type iEventName interface {
 	EventID() string //
 }
 
+// if developers implement this interface, we're spinning a goroutine if the event says it is async
+type iAsync interface {
+	Async() bool
+}
+
 // Listener is being returned when you subscribe to a topic, so you can unsubscribe or access the parent topic
 type Listener[T any] struct {
 	parent   *Topic[T]     // so we can call unsubscribe from parent
@@ -90,17 +95,32 @@ func (s *Listener[T]) Topic() *Topic[T] {
 // Pub allows you to publish an event in that topic
 func (b *Topic[T]) Pub(event T) {
 	b.rwMu.RLock()
-	for topic := range b.subs {
-		b.subs[topic].callback(event)
+
+	isAsync := false
+	switch m := any(event).(type) {
+	case iAsync:
+		isAsync = m.Async()
 	}
+
+	for sub := range b.subs {
+		if isAsync {
+			go b.subs[sub].callback(event)
+			continue
+		}
+
+		b.subs[sub].callback(event)
+	}
+
 	b.rwMu.RUnlock()
 }
 
 func (b *Topic[T]) PubAsync(event T) {
 	b.rwMu.RLock()
-	for topic := range b.subs {
-		go b.subs[topic].callback(event)
+
+	for sub := range b.subs {
+		go b.subs[sub].callback(event)
 	}
+
 	b.rwMu.RUnlock()
 }
 
@@ -121,7 +141,7 @@ func (o *Bus[T]) Unsub() {
 func SubUnsub[T any](callback func(event T) bool) *Bus[T] {
 	var (
 		event T
-		key string
+		key   string
 	)
 
 	switch m := any(event).(type) {
@@ -157,7 +177,7 @@ func SubUnsub[T any](callback func(event T) bool) *Bus[T] {
 func Sub[T any](callback func(event T)) *Bus[T] {
 	var (
 		event T
-		key string
+		key   string
 	)
 
 	switch m := any(event).(type) {
@@ -206,6 +226,7 @@ func Pub[T any](event T) {
 // PubAsync publishes an event which will be dispatched to all listeners
 func PubAsync[T any](event T) {
 	var key string
+
 	switch m := any(event).(type) {
 	case iEventName:
 		key = m.EventID()

main_test.go

@@ -1,6 +1,7 @@
 package bus_test
 
 import (
+	"sync"
 	"sync/atomic"
 	"testing"
 
@@ -24,7 +25,7 @@ func TestSubTopicWhilePub(t *testing.T) {
 	go func() {
 		<-start
 		topic.PubAsync(&Uint32AsyncEvent{u: 1})
-		close(finishPubWait)
+		defer close(finishPubWait)
 	}()
 
 	newSubCalled := false
@@ -42,6 +43,7 @@ func TestSubTopicWhilePub(t *testing.T) {
 	<-finishPubWait // wait for pub to finish
 
 	<-finishSubWait // wait for sub to finish
+
 	if newSubCalled {
 		t.Fatal("new subscriber should not be called")
 	}
@@ -78,3 +80,45 @@ func TestReusePayloadPointerAsync(t *testing.T) {
 
 	t.Logf("altered payload %d for %d listeners", payload.u, c)
 }
+
+func TestAsyncBus(t *testing.T) {
+	c := uint32(0)
+
+	var wg sync.WaitGroup
+	wg.Add(4096)
+	bus.Sub(
+		func(event Uint32AsyncEvent) {
+			atomic.AddUint32(&c, 1)
+			wg.Done()
+		},
+	)
+
+	go func() {
+		for i := 0; i < 1024; i++ {
+			bus.PubAsync(Uint32AsyncEvent{})
+		}
+	}()
+	go func() {
+		for i := 0; i < 1024; i++ {
+			bus.PubAsync(Uint32AsyncEvent{})
+		}
+	}()
+	go func() {
+		for i := 0; i < 1024; i++ {
+			bus.PubAsync(Uint32AsyncEvent{})
+		}
+	}()
+	go func() {
+		for i := 0; i < 1024; i++ {
+			bus.PubAsync(Uint32AsyncEvent{})
+		}
+	}()
+
+	wg.Wait()
+
+	if c != 4096 {
+		t.Fatalf("error : counter should be 4096 but is %d", c)
+	}
+
+	t.Logf("%d", c)
+}