introduce SubscriptionRef, add example to README

Signed-off-by: tomasz.ziolkowski <tomasz.ziolkowski@allegro.pl>

Author: ziollek

README.md

@@ -26,6 +26,9 @@ import (
 ```
 
 #### Example
+
+General purpose event bus (unlimited number of parameters of any type):
+
 ```go
 func calculator(a int, b int) {
 	fmt.Printf("%d\n", a + b)
@@ -39,6 +42,27 @@ func main() {
 }
 ```
 
+You can alternatively use ~50% faster implementation if your listener accepts only one argument of a particular type.
+Simple event bus (only one parameter of a chosen type):
+
+```go
+type sumComponents struct {
+    a int
+    b int
+}
+
+func calculator(args sumComponents) {
+	fmt.Printf("%d\n", args.a + args.b)
+}
+
+func main() {
+    bus := EventBus.NewSimpleBus[sumComponents]()
+    ref := bus.Subscribe("main:calculator", calculator)
+    bus.Publish("main:calculator", sumComponents{20, 40})
+    bus.Unsubscribe("main:calculator", ref)
+}
+```
+
 #### Implemented methods
 * **New()**
 * **Subscribe()**
@@ -56,10 +80,10 @@ New returns new EventBus with empty handlers.
 bus := EventBus.New();
 ```
 
-You can alternatively use ~50% faster implementation if your listener accepts only one argument of a particular type.
+#### NewSimpleBus[T]()
 The below example creates a new EventBus with handlers accepting only string arguments:
 ```go
-bus := EventBus.NewSimpleBus[string]();
+bus := EventBus.NewSimpleBus[string]()
 ````
 
 #### Subscribe(topic string, fn interface{}) error
@@ -70,6 +94,18 @@ func Handler() { ... }
 bus.Subscribe("topic:handler", Handler)
 ```
 
+#### Subscribe(topic string, fn func(T)) SubscriptionRef
+
+Simplified version of bus returns `SubscriptionRef` as a result of all subscription methods. It can be used to unsubscribe from the topic.
+```go
+func Handler(param string) { ... }
+...
+bus := EventBus.NewSimpleBus[string]()
+ref := bus.Subscribe("topic:handler", Handler)
+...
+bus.Unsubscribe(ref)
+```
+
 #### SubscribeOnce(topic string, fn interface{}) error
 Subscribe to a topic once. Handler will be removed after executing. Returns error if `fn` is not a function.
 ```go

generic_bus.go

@@ -1,17 +1,18 @@
 package EventBus
 
 import (
-	"reflect"
 	"sync"
 )
 
+type SubscriptionRef uint64
+
 // SimpleBusSubscriber defines subscription-related bus behavior for event of specific type T
 type SimpleBusSubscriber[T any] interface {
-	Subscribe(topic string, fn func(T))
-	SubscribeAsync(topic string, fn func(T), transactional bool)
-	SubscribeOnce(topic string, fn func(T))
-	SubscribeOnceAsync(topic string, fn func(T))
-	Unsubscribe(topic string, handler func(T))
+	Subscribe(topic string, fn func(T)) SubscriptionRef
+	SubscribeAsync(topic string, fn func(T), transactional bool) SubscriptionRef
+	SubscribeOnce(topic string, fn func(T)) SubscriptionRef
+	SubscribeOnceAsync(topic string, fn func(T)) SubscriptionRef
+	Unsubscribe(topic string, ref SubscriptionRef)
 }
 
 // SimpleBusPublisher defines publishing-related bus behavior for event of specific type T
@@ -26,16 +27,12 @@ type SimpleBus[T any] interface {
 	SimpleBusPublisher[T]
 }
 
-type callBack[T any] struct {
-	Call    func(T)
-	pointer uintptr
-}
-
 type genericHandler[T any] struct {
-	callBack      callBack[T]
+	callBack      func(T)
 	once          bool
 	async         bool
 	transactional bool
+	reference     SubscriptionRef
 	sync.Mutex    // lock for an event handler - useful for running async callbacks serially
 }
 
@@ -48,37 +45,28 @@ func (h *genericHandler[T]) Call(arg T, wg *sync.WaitGroup) {
 				h.Lock()
 				defer h.Unlock()
 			}
-			h.callBack.Call(arg)
+			h.callBack(arg)
 		}()
 		return
 	} else {
-		h.callBack.Call(arg)
+		h.callBack(arg)
 	}
 }
 
-type listeners[T any] []*genericHandler[T]
+type listeners[T any] map[SubscriptionRef]*genericHandler[T]
 
-func (l *listeners[T]) add(handler *genericHandler[T]) {
-	for _, h := range *l {
-		if h.callBack.pointer == handler.callBack.pointer {
-			return
-		}
-	}
-	*l = append(*l, handler)
+func (l listeners[T]) add(handler *genericHandler[T]) {
+	l[handler.reference] = handler
 }
 
-func (l *listeners[T]) delete(handler *genericHandler[T]) {
-	for i, h := range *l {
-		if h.callBack.pointer == handler.callBack.pointer {
-			*l = append((*l)[:i], (*l)[i+1:]...)
-			return
-		}
-	}
+func (l listeners[_]) delete(ref SubscriptionRef) {
+	delete(l, ref)
 }
 
-func newGenericHandler[T any](fn func(T), async, transactional, once bool) *genericHandler[T] {
+func newGenericHandler[T any](fn func(T), ref SubscriptionRef, async, transactional, once bool) *genericHandler[T] {
 	return &genericHandler[T]{
-		callBack:      callBack[T]{Call: fn, pointer: reflect.ValueOf(fn).Pointer()},
+		callBack:      fn,
+		reference:     ref,
 		async:         async,
 		transactional: transactional,
 		once:          once,
@@ -87,13 +75,15 @@ func newGenericHandler[T any](fn func(T), async, transactional, once bool) *gene
 
 type GenericBus[T any] struct {
 	handlers map[string]*listeners[T]
+	lastRef  SubscriptionRef
 	sync.RWMutex
 	wg sync.WaitGroup
 }
 
 func NewSimpleBus[T any]() SimpleBus[T] {
 	return &GenericBus[T]{
 		handlers: make(map[string]*listeners[T]),
+		lastRef:  SubscriptionRef(0),
 	}
 }
 
@@ -104,27 +94,27 @@ func (bus *GenericBus[_]) HasCallback(topic string) bool {
 	return ok && len(*bus.handlers[topic]) > 0
 }
 
-func (bus *GenericBus[T]) Subscribe(topic string, fn func(T)) {
-	bus.subscribe(topic, fn, false, false, false)
+func (bus *GenericBus[T]) Subscribe(topic string, fn func(T)) SubscriptionRef {
+	return bus.subscribe(topic, fn, false, false, false)
 }
 
-func (bus *GenericBus[T]) SubscribeOnce(topic string, fn func(T)) {
-	bus.subscribe(topic, fn, false, false, true)
+func (bus *GenericBus[T]) SubscribeOnce(topic string, fn func(T)) SubscriptionRef {
+	return bus.subscribe(topic, fn, false, false, true)
 }
 
-func (bus *GenericBus[T]) SubscribeOnceAsync(topic string, fn func(T)) {
-	bus.subscribe(topic, fn, true, false, true)
+func (bus *GenericBus[T]) SubscribeOnceAsync(topic string, fn func(T)) SubscriptionRef {
+	return bus.subscribe(topic, fn, true, false, true)
 }
 
-func (bus *GenericBus[T]) SubscribeAsync(topic string, fn func(T), transactional bool) {
-	bus.subscribe(topic, fn, true, transactional, false)
+func (bus *GenericBus[T]) SubscribeAsync(topic string, fn func(T), transactional bool) SubscriptionRef {
+	return bus.subscribe(topic, fn, true, transactional, false)
 }
 
-func (bus *GenericBus[T]) Unsubscribe(topic string, fn func(T)) {
+func (bus *GenericBus[T]) Unsubscribe(topic string, ref SubscriptionRef) {
 	bus.Lock()
 	defer bus.Unlock()
 	if _, ok := bus.handlers[topic]; ok {
-		bus.handlers[topic].delete(newGenericHandler(fn, false, false, false))
+		bus.handlers[topic].delete(ref)
 	}
 }
 
@@ -135,7 +125,7 @@ func (bus *GenericBus[T]) Publish(topic string, arg T) {
 		for _, subscriber := range *subscribers {
 			fire <- subscriber
 			if subscriber.once {
-				bus.handlers[topic].delete(subscriber)
+				bus.handlers[topic].delete(subscriber.reference)
 			}
 		}
 		bus.Unlock()
@@ -151,13 +141,15 @@ func (bus *GenericBus[T]) WaitAsync() {
 	bus.wg.Wait()
 }
 
-func (bus *GenericBus[T]) subscribe(topic string, fn func(T), async, transactional, once bool) {
+func (bus *GenericBus[T]) subscribe(topic string, fn func(T), async, transactional, once bool) SubscriptionRef {
 	bus.Lock()
 	defer bus.Unlock()
+	bus.lastRef++
 	if _, ok := bus.handlers[topic]; !ok {
 		bus.handlers[topic] = &listeners[T]{}
 	}
-	bus.handlers[topic].add(newGenericHandler(fn, async, transactional, once))
+	bus.handlers[topic].add(newGenericHandler(fn, bus.lastRef, async, transactional, once))
+	return bus.lastRef
 }
 
 func (bus *GenericBus[T]) fetchSubscribers(topic string) (*listeners[T], bool) {

generic_bus_test.go

@@ -1,6 +1,7 @@
 package EventBus
 
 import (
+	"fmt"
 	"testing"
 	"time"
 )
@@ -28,14 +29,17 @@ func TestSimpleOnceAndManySubscribe(t *testing.T) {
 	bus := NewSimpleBus[any]()
 	topic := "topic"
 	flag := 0
-	// artificially define different functions with the same logic
-	// subscription logic prevents from subscribe the same callback for the same topic multiple times
-	fn1 := func(_ any) { flag += 1 }
-	fn2 := func(_ any) { flag += 1 }
-	fn3 := func(_ any) { flag += 1 }
-	bus.SubscribeOnce(topic, fn1)
-	bus.Subscribe(topic, fn2)
-	bus.Subscribe(topic, fn3)
+
+	fn := func(_ any) { flag += 1 }
+
+	refFirst := bus.SubscribeOnce(topic, fn)
+	refSecond := bus.Subscribe(topic, fn)
+	refThird := bus.Subscribe(topic, fn)
+
+	if refFirst == refSecond || refFirst == refThird || refSecond == refThird {
+		t.Fail()
+	}
+
 	bus.Publish(topic, nil)
 
 	if flag != 3 {
@@ -46,12 +50,12 @@ func TestSimpleOnceAndManySubscribe(t *testing.T) {
 func TestSimpleUnsubscribe(t *testing.T) {
 	bus := NewSimpleBus[any]()
 	fn := func(_ any) {}
-	bus.Subscribe("topic", fn)
+	ref := bus.Subscribe("topic", fn)
 	if !bus.HasCallback("topic") {
 		t.Logf("Expected to have callback for topic but it is not present")
 		t.Fail()
 	}
-	bus.Unsubscribe("topic", fn)
+	bus.Unsubscribe("topic", ref)
 	if bus.HasCallback("topic") {
 		t.Logf("Expected to have no callback for topic after unsubscribe but it is present")
 		t.Fail()
@@ -70,9 +74,9 @@ func TestSimpleUnsubscribeMethod(t *testing.T) {
 	bus := NewSimpleBus[any]()
 	h := &accumulator{val: 0}
 
-	bus.Subscribe("topic", h.Handle)
+	ref := bus.Subscribe("topic", h.Handle)
 	bus.Publish("topic", nil)
-	bus.Unsubscribe("topic", h.Handle)
+	bus.Unsubscribe("topic", ref)
 	bus.Publish("topic", nil)
 
 	if h.val != 1 {
@@ -218,3 +222,19 @@ func BenchmarkSimpleAsynchronousPublishing(b *testing.B) {
 	}
 	bus.WaitAsync()
 }
+
+type sumComponents struct {
+	a int
+	b int
+}
+
+func calculator(args sumComponents) {
+	fmt.Printf("%d\n", args.a+args.b)
+}
+
+func TestSimpleExample(t *testing.T) {
+	bus := NewSimpleBus[sumComponents]()
+	ref := bus.Subscribe("main:calculator", calculator)
+	bus.Publish("main:calculator", sumComponents{20, 40})
+	bus.Unsubscribe("main:calculator", ref)
+}