Fix closing event bus

.golangci.yml

@@ -24,6 +24,7 @@ linters:
     - sqlclosecheck # disabled because of generics
     - tagliatelle # requires ID instead of Id in jsonTags, which is annoying for our old structs
     - varcheck  # deprecated
+    - perfsprint
 linters-settings:
   cyclop:
     max-complexity: 20

eventbus.go

@@ -169,11 +169,13 @@ func (b *EventBus) AddHandler(ctx context.Context, matcher eh.EventMatcher, hand
 		return eh.ErrMissingHandler
 	}
 
+	handlerType := handler.HandlerType()
+
 	// Check handler existence.
 	b.registeredMu.Lock()
 	defer b.registeredMu.Unlock()
 
-	if _, ok := b.registered[handler.HandlerType()]; ok {
+	if _, ok := b.registered[handlerType]; ok {
 		return eh.ErrHandlerAlreadyAdded
 	}
 
@@ -184,10 +186,10 @@ func (b *EventBus) AddHandler(ctx context.Context, matcher eh.EventMatcher, hand
 
 	// Handle until context is cancelled.
 	b.wg.Add(1)
-	go b.handle(consumer)
+	go b.handleCancel(handlerType)
 
 	// Register handler.
-	b.registered[handler.HandlerType()] = consumer
+	b.registered[handlerType] = consumer
 
 	return nil
 }

go.mod

@@ -3,7 +3,7 @@ module github.com/Clarilab/eh-rabbitmq
 go 1.21
 
 require (
-	github.com/Clarilab/clarimq v1.3.0
+	github.com/Clarilab/clarimq v1.3.1
 	github.com/Clarilab/eh-tracygo v1.0.0
 	github.com/Clarilab/tracygo/v2 v2.1.0
 	github.com/google/uuid v1.6.0

go.sum

@@ -1,5 +1,5 @@
-github.com/Clarilab/clarimq v1.3.0 h1:+xUz8NVIa0x/clEtfYp1ZWGLp3HDT2jfy8XFN5G1ZYo=
-github.com/Clarilab/clarimq v1.3.0/go.mod h1:xMNpFq9qb6MvssW9H3567qeO1ohfnrVedXJ50z27NsM=
+github.com/Clarilab/clarimq v1.3.1 h1:RDDZZ+8/X3IWdPhVzo0RKn5Ia2pM71KGT1bed6GEsII=
+github.com/Clarilab/clarimq v1.3.1/go.mod h1:xMNpFq9qb6MvssW9H3567qeO1ohfnrVedXJ50z27NsM=
 github.com/Clarilab/eh-tracygo v1.0.0 h1:2KlaQfPO5ajqBinHK4Hu3vXauTG2q0W9jJ7Aim5pQFw=
 github.com/Clarilab/eh-tracygo v1.0.0/go.mod h1:bC1/XWv6byi0+YzkvuoioeV8vPruyK0jHkFo5gN5+Qc=
 github.com/Clarilab/tracygo/v2 v2.1.0 h1:5QU3NTTboXD9PK0RCvSe7hBnO/vafkX05S2IYe3m44s=

handlers.go

@@ -13,17 +13,24 @@ const (
 	retryCountHeader = "x-retry-count"
 )
 
-// Handles all events coming in on the channel.
-func (b *EventBus) handle(
-	consumer *clarimq.Consumer,
+func (b *EventBus) handleCancel(
+	handlerType eh.EventHandlerType,
 ) {
 	defer b.wg.Done()
 
 	<-b.ctx.Done()
 
-	b.consumerMu.Lock()
-	consumer.Close()
-	b.consumerMu.Unlock()
+	b.registeredMu.RLock()
+	defer b.registeredMu.RUnlock()
+
+	if consumer, ok := b.registered[handlerType]; ok {
+		b.consumerMu.Lock()
+		defer b.consumerMu.Unlock()
+
+		if err := consumer.Close(); err != nil {
+			b.logger.logInfo(fmt.Sprintf("failed to close consumer for handler: %s", handlerType.String()))
+		}
+	}
 }
 
 func (b *EventBus) handler(