Fixes and notes

Author: badu

README.md

@@ -26,6 +26,8 @@ The event producer will simply do:
 
 `bus.Pub(InterestingEvent{})`
 
+Usage : `go get github.com/badu/bus`
+
 ## What Problem Does It Solve?
 
 Decoupling of components: publishers and subscribers can operate independently of each other, with no direct knowledge
@@ -47,18 +49,77 @@ Each component can then be developed and tested independently, making the overal
 
 Inside the `test_scenarios` folder, you can find the following scenarios:
 
-1. Fire and Forget
+1. Fire and Forget.
+
+   Imagine a system / application where we have three services : `users`, `notifications` (email and
+   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)
+   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
+   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.
+
+   Simulating audit service malfunction is easy. Instead of using `Sub`, we're using `SubUnsub` to register the listener
+   and return [`true`](https://github.com/badu/bus/blob/main/test_scenarios/fire-and-forget/audit/service.go#L36) to
+   unsubscribe on events of that kind.
+
+2. Factory Request Reply
+
+   Imagine a system / application where we need to communicate with different microservices, but in this case we don't
+   want to bring them online, we're just wanting to stub the response as those services were alive.
+
+   This technique is useful when we need to test some complicated flows of business logic and facilitates the
+   transformation of an integration test into a classic unit test.
+
+   The `cart` service requires two replies from two other microservices `inventory` and `prices`. In the past, I've been
+   using a closure function to provide the service with both real GRPC clients or with mocks and stubs. The service
+   signature gets complicated and large as we one service would depend on a lot of GRPC clients to aggregate data.
+
+   As you can see
+   the [test here](https://github.com/badu/bus/blob/main/test_scenarios/factory-request-reply/main_test.go) it's much
+   more elegant and the service constructor is much slimmer.
+
+   Events are one sync and one async, just to check it works in both scenarios.
+
+   Important to note that because a `WaitGroup` is being used in our event struct, we're forced to pass the events by
+   using a pointer, instead of passing them by value.
+
+3. Request Reply with Callback
+
+   In this example, we wanted to achieve two things. First is that the `service` and the `repository` are decoupled by
+   events. More than that, we wanted that the events are generic on their own.
+
+   The orders service will dispatch a generic request event, one for placing an order, which will carry an `Order` (
+   model) struct with that request and another `OrderStatus` (model) struct using the same generic event.
+
+   We are using a channel inside the generic `RequestEvent` to signal the `reply` to the publisher, which in this case
+   is a callback function that returns the result as if the publisher would have called directly the listener.
+
+   I am sure that you will find this technique interesting and having a large number of applications.
+
+   An important note is about not forgetting to implement the `EventID() string` correctly, as incorrect naming triggers
+   panic (expecting one type of event, but receiving another). To exemplify this, just alter the return of
+   this [function](https://github.com/badu/bus/blob/main/test_scenarios/request-reply-callback/events/main.go#L24).
 
-2. Request Reply with Callback
+4. Request Reply with Cancellation
 
-3. Request Reply with Cancellation
+   Last but, not least, this is an example about providing `context.Context` along the publisher subscriber chain.
+   The `repository` is simulating a long database call, longer than the context's cancellation, so the service gets the
+   deadline exceeded error.
 
-4. Factory Request Reply
+   Note that this final example is not using pointer to the event's struct, but it contains two properties which have
+   pointers, so the `service` can access the altered `reply`.
 
 ## Recommendations
 
-1. when using `sync.WaitGroup` inside your event struct, always use method receivers and pass the event as pointer,
-   otherwise you will be passing a lock by value (which is `sync.Locker`).
+1. always place your events inside a separate `events` package, avoiding circular dependencies.
+2. in general, in `request-reply` scenarios, the events should be passed as pointers (even if it's somewhat slower),
+   because changing properties that represents the `reply` would not be reflected. Also, when using `sync.WaitGroup`
+   inside your event struct, always use method receivers and pass the event as pointer, otherwise you will be passing a
+   lock by value (which is `sync.Locker`).
 2. be careful if you don't want to use pointers for events, but you still need to pass values from the listener to the
    dispatcher. You should still have at least one property of that event that is a pointer (see events
    in `request reply with cancellation` for example). Same technique can be applied when you need `sync.Waitgroup` to be

main_test.go

@@ -48,8 +48,7 @@ func TestSubTopicWhilePub(t *testing.T) {
 }
 
 func TestReusePayloadPointerAsync(t *testing.T) {
-	// if you reuse the payload, you can alter it's content
-	// which is not recommended (see the "random" number of goroutines that catches that change)
+	// if you reuse the payload, you can alter it's content, of course
 
 	topic := bus.NewTopic[*Uint32AsyncEvent]()
 	c := uint32(0)

test_scenarios/fire-and-forget/audit/service.go

@@ -14,10 +14,23 @@ type ServiceImpl struct {
 
 func NewAuditService(sb *strings.Builder) ServiceImpl {
 	result := ServiceImpl{sb: sb}
+	bus.Sub(result.OnUserRegisteredEvent)
+	bus.SubUnsub(result.OnSMSRequestEvent)
 	bus.SubUnsub(result.OnSMSSentEvent)
 	return result
 }
 
+// OnUserRegisteredEvent is classic event handler
+func (s *ServiceImpl) OnUserRegisteredEvent(event events.UserRegisteredEvent) {
+	// we can save audit data here
+}
+
+// OnSMSRequestEvent is a pub-unsub type, we have to return 'false' to continue listening for this kind of events
+func (s *ServiceImpl) OnSMSRequestEvent(event events.SMSRequestEvent) bool {
+	return false
+}
+
+// OnSMSSentEvent is a pub-unsub type where we give up on listening after receiving first message
 func (s *ServiceImpl) OnSMSSentEvent(event events.SMSSentEvent) bool {
 	s.sb.WriteString(fmt.Sprintf("audit event : an sms was %s sent to %s with message %s\n", event.Status, event.Request.Number, event.Request.Message))
 	return true // after first event, audit will give up listening for events

test_scenarios/request-reply-callback/events/main.go

@@ -1,5 +1,9 @@
 package events
 
+import (
+	"fmt"
+)
+
 const RequestEventType = "RequestEvent"
 
 type RequestEvent[T any] struct {
@@ -8,8 +12,16 @@ type RequestEvent[T any] struct {
 	Done     chan struct{}
 }
 
+func NewRequestEvent[T any](payload T) *RequestEvent[T] {
+	return &RequestEvent[T]{
+		Payload: payload,
+		Done:    make(chan struct{}),
+	}
+}
+
 func (i *RequestEvent[T]) EventID() string {
-	return RequestEventType
+	var t T
+	return fmt.Sprintf("%s%T", RequestEventType, t)
 }
 
 func (i *RequestEvent[T]) Async() bool {

test_scenarios/request-reply-callback/main_test.go

@@ -10,6 +10,7 @@ import (
 
 func TestRequestReplyCallback(t *testing.T) {
 	var sb strings.Builder
+	orders.NewRepository(&sb)
 	svc := orders.NewService(&sb)
 
 	ctx := context.Background()
@@ -53,4 +54,6 @@ func TestRequestReplyCallback(t *testing.T) {
 	}
 	t.Logf("order #2 status : %s", stat2.Status)
 
+	t.Logf("%s", sb.String())
+
 }

test_scenarios/request-reply-callback/orders/repository.go

@@ -23,14 +23,10 @@ type RepositoryImpl struct {
 	calls int
 }
 
-func NewRepository(
-	sb *strings.Builder,
-	cBus *bus.Topic[*events.RequestEvent[Order]],
-	sBus *bus.Topic[*events.RequestEvent[OrderStatus]],
-) RepositoryImpl {
+func NewRepository(sb *strings.Builder) RepositoryImpl {
 	result := RepositoryImpl{sb: sb}
-	cBus.Sub(result.onCreateOrder)
-	sBus.Sub(result.onGetOrderStatus)
+	bus.Sub(result.onCreateOrder)
+	bus.Sub(result.onGetOrderStatus)
 	return result
 }
 

test_scenarios/request-reply-callback/orders/service.go

@@ -2,36 +2,34 @@ package orders
 
 import (
 	"context"
+	"fmt"
 	"strings"
 
 	"github.com/badu/bus"
 	"github.com/badu/bus/test_scenarios/request-reply-callback/events"
 )
 
 type ServiceImpl struct {
-	sb   *strings.Builder
-	CBus *bus.Topic[*events.RequestEvent[Order]]
-	SBus *bus.Topic[*events.RequestEvent[OrderStatus]]
+	sb *strings.Builder
 }
 
 func NewService(sb *strings.Builder) ServiceImpl {
 	result := ServiceImpl{sb: sb}
-	result.CBus = bus.NewTopic[*events.RequestEvent[Order]]()
-	result.SBus = bus.NewTopic[*events.RequestEvent[OrderStatus]]()
-	NewRepository(sb, result.CBus, result.SBus)
 	return result
 }
 
 func (s *ServiceImpl) RegisterOrder(ctx context.Context, productIDs []int) (*Order, error) {
-	event := events.RequestEvent[Order]{Payload: Order{ProductIDs: productIDs}, Done: make(chan struct{})}
-	s.CBus.Pub(&event)
-	<-event.Done
-	return event.Callback()
+	event := events.NewRequestEvent[Order](Order{ProductIDs: productIDs})
+	s.sb.WriteString(fmt.Sprintf("dispatching event typed %s\n", event.EventID()))
+	bus.Pub(event)
+	<-event.Done            // wait for "reply"
+	return event.Callback() // return the callback, which is containing the actual result
 }
 
 func (s *ServiceImpl) GetOrderStatus(ctx context.Context, orderID int) (*OrderStatus, error) {
-	event := events.RequestEvent[OrderStatus]{Payload: OrderStatus{OrderID: orderID}, Done: make(chan struct{})}
-	s.SBus.Pub(&event)
-	<-event.Done
-	return event.Callback()
+	event := events.NewRequestEvent[OrderStatus](OrderStatus{OrderID: orderID})
+	s.sb.WriteString(fmt.Sprintf("dispatching event typed %s\n", event.EventID()))
+	bus.Pub(event)
+	<-event.Done            // wait for "reply"
+	return event.Callback() // return the callback, which is containing the actual result
 }

test_scenarios/request-reply-with-cancellation/events/main.go

@@ -22,8 +22,12 @@ func (s *EventState) Close() {
 	close(s.Done)
 }
 
+type NewOrder struct {
+	ID int
+}
+
 type CreateOrderEvent struct {
-	OrderID    int
+	NewOrder   *NewOrder
 	ProductIDs []int
 	State      *EventState
 }

test_scenarios/request-reply-with-cancellation/main_test.go

@@ -5,16 +5,13 @@ import (
 	"testing"
 	"time"
 
-	"github.com/badu/bus"
-	"github.com/badu/bus/test_scenarios/request-reply-with-cancellation/events"
 	"github.com/badu/bus/test_scenarios/request-reply-with-cancellation/orders"
 )
 
 func TestRequestReplyWithCancellation(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*1)
-	ebus := bus.NewTopic[events.CreateOrderEvent]()
-	svc := orders.NewService(ebus)
-	orders.NewRepository(ebus)
+	svc := orders.NewService()
+	orders.NewRepository()
 
 	response, err := svc.CreateOrder(ctx, []int{1, 2, 3})
 	switch err {

test_scenarios/request-reply-with-cancellation/orders/repository.go

@@ -16,7 +16,7 @@ type RepositoryImpl struct {
 	calls int
 }
 
-func NewRepository(bus *bus.Topic[events.CreateOrderEvent]) RepositoryImpl {
+func NewRepository() RepositoryImpl {
 	result := RepositoryImpl{}
 	bus.Sub(result.OnCreateOrder)
 	return result
@@ -30,7 +30,7 @@ func (r *RepositoryImpl) OnCreateOrder(event events.CreateOrderEvent) {
 	for {
 		select {
 		case <-time.After(4 * time.Second):
-			event.OrderID = r.calls
+			event.NewOrder = &events.NewOrder{ID: r.calls}
 			event.State.Close()
 			return
 		case <-event.State.Ctx.Done():