added locks for middlewares in router (#465)

Co-authored-by: Lara Blyshchyk <[email protected]> Co-authored-by: Miłosz Smółka <[email protected]>
ThreeDotsLabs · Aug 23, 2024 · b7676e5 · b7676e5
1 parent e0c6f18
commit b7676e5
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 3 deletions.
diff --git a/_examples/pubsubs/googlecloud/go.sum b/_examples/pubsubs/googlecloud/go.sum
@@ -137,6 +137,7 @@ google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98
 google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
 google.golang.org/genproto v0.0.0-20230202175211-008b39050e57 h1:vArvWooPH749rNHpBGgVl+U9B9dATjiEhJzcWGlovNs=
 google.golang.org/genproto v0.0.0-20230202175211-008b39050e57/go.mod h1:RGgjbofJ8xD9Sq1VVhDM1Vok1vRONV+rg+CjzG4SZKM=
+google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.21.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
 google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=

diff --git a/message/router.go b/message/router.go
@@ -107,7 +107,8 @@ func NewRouter(config RouterConfig, logger watermill.LoggerAdapter) (*Router, er
 
 		handlerAdded: make(chan struct{}),
 
-		handlersLock: &sync.RWMutex{},
+		middlewaresLock: &sync.RWMutex{},
+		handlersLock:    &sync.RWMutex{},
 
 		closingInProgressCh: make(chan struct{}),
 		closedCh:            make(chan struct{}),
@@ -131,7 +132,8 @@ type middleware struct {
 type Router struct {
 	config RouterConfig
 
-	middlewares []middleware
+	middlewares     []middleware
+	middlewaresLock *sync.RWMutex
 
 	plugins []RouterPlugin
 
@@ -185,6 +187,8 @@ func (r *Router) addRouterLevelMiddleware(m ...HandlerMiddleware) {
 }
 
 func (r *Router) addHandlerLevelMiddleware(handlerName string, m ...HandlerMiddleware) {
+	r.middlewaresLock.Lock()
+	defer r.middlewaresLock.Unlock()
 	for _, handlerMiddleware := range m {
 		middleware := middleware{
 			Handler:       handlerMiddleware,
@@ -435,7 +439,11 @@ func (r *Router) RunHandlers(ctx context.Context) error {
 		go func() {
 			defer cancel()
 
-			h.run(ctx, r.middlewares)
+			r.middlewaresLock.Lock()
+			middlewares := append([]middleware{}, r.middlewares...)
+			r.middlewaresLock.Unlock()
+
+			h.run(ctx, middlewares)
 
 			r.handlersWg.Done()
 			r.logger.Info("Subscriber stopped", watermill.LogFields{