feat: ✨ add thread-safe handlers registration

Author: mehdihadeli

go.mod

@@ -3,7 +3,6 @@ module github.com/mehdihadeli/go-mediatr
 go 1.24
 
 require (
-	github.com/goccy/go-reflect v1.2.0
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.10.0
 )

internal/examples/cqrs_example/go.mod

@@ -1,6 +1,8 @@
 module cqrsexample
 
-go 1.18
+go 1.24
+
+toolchain go1.24.2
 
 replace github.com/mehdihadeli/go-mediatr => ../../../
 

mediatr.go

@@ -2,8 +2,9 @@ package mediatr
 
 import (
 	"context"
+	"reflect"
+	"sync"
 
-	"github.com/goccy/go-reflect"
 	"github.com/pkg/errors"
 )
 
@@ -27,16 +28,22 @@ type NotificationHandler[TNotification any] interface {
 
 type NotificationHandlerFactory[TNotification any] func() NotificationHandler[TNotification]
 
-var requestHandlersRegistrations = map[reflect.Type]interface{}{}
-var notificationHandlersRegistrations = map[reflect.Type][]interface{}{}
-var pipelineBehaviours []interface{} = []interface{}{}
+var (
+	requestHandlersRegistrations      = map[reflect.Type]interface{}{}
+	notificationHandlersRegistrations = map[reflect.Type][]interface{}{}
+	pipelineBehaviours                []interface{}
+	registryMutex                     sync.RWMutex
+)
 
 type Unit struct{}
 
 func registerRequestHandler[TRequest any, TResponse any](handler any) error {
 	var request TRequest
 	requestType := reflect.TypeOf(request)
 
+	registryMutex.Lock()
+	defer registryMutex.Unlock()
+
 	_, exist := requestHandlersRegistrations[requestType]
 	if exist {
 		// each request in request/response strategy should have just one handler
@@ -60,6 +67,9 @@ func RegisterRequestHandlerFactory[TRequest any, TResponse any](factory RequestH
 
 // RegisterRequestPipelineBehaviors register the request behaviors to mediatr registry.
 func RegisterRequestPipelineBehaviors(behaviours ...PipelineBehavior) error {
+	registryMutex.Lock()
+	defer registryMutex.Unlock()
+
 	for _, behavior := range behaviours {
 		behaviorType := reflect.TypeOf(behavior)
 
@@ -78,6 +88,9 @@ func registerNotificationHandler[TEvent any](handler any) error {
 	var event TEvent
 	eventType := reflect.TypeOf(event)
 
+	registryMutex.Lock()
+	defer registryMutex.Unlock()
+
 	handlers, exist := notificationHandlersRegistrations[eventType]
 	if !exist {
 		notificationHandlersRegistrations[eventType] = []interface{}{handler}
@@ -157,7 +170,15 @@ func buildRequestHandler[TRequest any, TResponse any](handler any) (RequestHandl
 func Send[TRequest any, TResponse any](ctx context.Context, request TRequest) (TResponse, error) {
 	requestType := reflect.TypeOf(request)
 	var response TResponse
+
+	registryMutex.RLock()
 	handler, ok := requestHandlersRegistrations[requestType]
+	// without copying, another goroutine could modify the original slice
+	behavioursCopy := make([]interface{}, len(pipelineBehaviours))
+	// deep copy of elements
+	copy(behavioursCopy, pipelineBehaviours)
+	registryMutex.RUnlock()
+
 	if !ok {
 		// request-response strategy should have exactly one handler and if we can't find a corresponding handler, we should return an error
 		return *new(TResponse), errors.Errorf("no handler for request %T", request)
@@ -168,29 +189,29 @@ func Send[TRequest any, TResponse any](ctx context.Context, request TRequest) (T
 		return *new(TResponse), errors.Errorf("handler for request %T is not a Handler", request)
 	}
 
-	if len(pipelineBehaviours) > 0 {
-		var reversPipes = reversOrder(pipelineBehaviours)
+	if len(behavioursCopy) > 0 {
+		var reversPipes = reversOrder(behavioursCopy)
 
 		var lastHandler RequestHandlerFunc = func(ctx context.Context) (interface{}, error) {
 			return handlerValue.Handle(ctx, request)
 		}
-		
-        aggregateResult := lastHandler
-        for _, pipe := range reversPipes {
-            pipeValue := pipe.(PipelineBehavior) 
-            currentNext := aggregateResult
-            
-            aggregateResult = func(ctx context.Context) (interface{}, error) {
-                return pipeValue.Handle(ctx, request, currentNext)
-            }
-        }
-
-        response, err := aggregateResult(ctx)
-        if err != nil {
-            return *new(TResponse), errors.Wrap(err, "error handling request")
-        }
-
-        return response.(TResponse), nil
+
+		aggregateResult := lastHandler
+		for _, pipe := range reversPipes {
+			pipeValue := pipe.(PipelineBehavior)
+			currentNext := aggregateResult
+
+			aggregateResult = func(ctx context.Context) (interface{}, error) {
+				return pipeValue.Handle(ctx, request, currentNext)
+			}
+		}
+
+		response, err := aggregateResult(ctx)
+		if err != nil {
+			return *new(TResponse), errors.Wrap(err, "error handling request")
+		}
+
+		return response.(TResponse), nil
 	} else {
 		res, err := handlerValue.Handle(ctx, request)
 		if err != nil {
@@ -221,13 +242,19 @@ func buildNotificationHandler[TNotification any](handler any) (NotificationHandl
 func Publish[TNotification any](ctx context.Context, notification TNotification) error {
 	eventType := reflect.TypeOf(notification)
 
+	registryMutex.RLock()
 	handlers, ok := notificationHandlersRegistrations[eventType]
+	// without copying, another goroutine could modify the original slice
+	handlersCopy := make([]interface{}, len(handlers))
+	// deep copy of elements
+	copy(handlersCopy, handlers)
+	registryMutex.RUnlock()
 	if !ok {
 		// notification strategy should have zero or more handlers, so it should run without any error if we can't find a corresponding handler
 		return nil
 	}
 
-	for _, handler := range handlers {
+	for _, handler := range handlersCopy {
 		handlerValue, ok := buildNotificationHandler[TNotification](handler)
 
 		if !ok {

mediatr_benchmarks_test.go

@@ -2,7 +2,7 @@ package mediatr
 
 import (
 	"context"
-	"github.com/goccy/go-reflect"
+	"reflect"
 	"testing"
 )
 

mediatr_test.go

@@ -3,15 +3,17 @@ package mediatr
 import (
 	"context"
 	"fmt"
+	"reflect"
+	"sync"
 	"testing"
 
-	"github.com/goccy/go-reflect"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 var testData []string
+var testMutex sync.Mutex
 
 func TestRunner(t *testing.T) {
 	//https://pkg.go.dev/testing@master#hdr-Subtests_and_Sub_benchmarks
@@ -186,20 +188,19 @@ func (t *MediatRTests) Test_Send_Should_Dispatch_Request_To_Handler_And_Get_Resp
 	pip1 := &PipelineBehaviourTest{}
 	pip2 := &PipelineBehaviourTest2{}
 	err := RegisterRequestPipelineBehaviors(pip1, pip2)
-	if err != nil {
-		t.Errorf("error registering request pipeline behaviors: %s", err)
-	}
+	assert.Nil(t, err)
 
 	handler := &RequestTestHandler{}
 	errRegister := RegisterRequestHandler[*RequestTest, *ResponseTest](handler)
-	if errRegister != nil {
-		t.Error(errRegister)
-	}
+	assert.Nil(t, errRegister)
 
 	response, err := Send[*RequestTest, *ResponseTest](context.Background(), &RequestTest{Data: "test"})
 	assert.Nil(t, err)
 	assert.IsType(t, &ResponseTest{}, response)
 	assert.Equal(t, "test", response.Data)
+
+	testMutex.Lock()
+	defer testMutex.Unlock()
 	assert.Contains(t, testData, "PipelineBehaviourTest")
 	assert.Contains(t, testData, "PipelineBehaviourTest2")
 }
@@ -328,7 +329,7 @@ func (t *MediatRTests) Test_Clear_Request_Registrations() {
 	err2 := RegisterRequestHandler[*RequestTest2, *ResponseTest2](handler2)
 	require.NoError(t, err1, err2)
 
-	ClearRequestRegistrations()
+	cleanup()
 
 	count := len(requestHandlersRegistrations)
 	assert.Equal(t, 0, count)
@@ -481,7 +482,14 @@ func (c *PipelineBehaviourTest2) Handle(ctx context.Context, request interface{}
 
 // /////////////////////////////////////////////////////////////////////////////////////////////
 func cleanup() {
-	requestHandlersRegistrations = map[reflect.Type]interface{}{}
-	notificationHandlersRegistrations = map[reflect.Type][]interface{}{}
-	pipelineBehaviours = []interface{}{}
+	testMutex.Lock()
+	defer testMutex.Unlock()
+
+	// Reset package-level registrations
+	requestHandlersRegistrations = make(map[reflect.Type]interface{})
+	notificationHandlersRegistrations = make(map[reflect.Type][]interface{})
+	pipelineBehaviours = make([]interface{}, 0)
+
+	// Reset test data
+	testData = nil
 }