Fix: delete publisher when producer is removed from manager

Signed-off-by: Aitor Perez Cedres <[email protected]>
rabbitmq · Dec 15, 2023 · 3dcf74d · 3dcf74d
1 parent 6ef17b7
commit 3dcf74d
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 0 deletions.
diff --git a/pkg/stream/producer_manager.go b/pkg/stream/producer_manager.go
@@ -135,6 +135,12 @@ func (p *producerManager) removeProducer(id int) error {
 	if id < 0 {
 		return errNegativeId
 	}
+	ctx, cancel := maybeApplyDefaultTimeout(context.Background())
+	defer cancel()
+	p.clientM.Lock()
+	_ = p.client.DeletePublisher(ctx, uint8(id))
+	// TODO log error if not nil
+	p.clientM.Unlock()
 	p.producers[id] = nil
 	p.producerCount -= 1
 

diff --git a/pkg/stream/producer_manager_test.go b/pkg/stream/producer_manager_test.go
@@ -95,6 +95,8 @@ var _ = Describe("ProducerManager", func() {
 			prepareMockDeclarePublisher(fakeRawClient, 1)
 			prepareMockDeclarePublisher(fakeRawClient, 1)
 			prepareMockDeclarePublisher(fakeRawClient, 2)
+			prepareMockDeletePublisher(fakeRawClient, 1)
+			prepareMockDeletePublisher(fakeRawClient, 2)
 			pm := newProducerManager(0, EnvironmentConfiguration{
 				MaxProducersByConnection: 5,
 			})
@@ -144,6 +146,7 @@ var _ = Describe("ProducerManager", func() {
 		It("closes the connection", func() {
 			fakeRawClient.EXPECT().Close(gomock.Any())
 			prepareMockDeclarePublisher(fakeRawClient, 0)
+			prepareMockDeletePublisher(fakeRawClient, 0) // called during producer.Close() callback
 			pm := newProducerManager(0, EnvironmentConfiguration{MaxProducersByConnection: 5})
 			pm.client = fakeRawClient
 			pm.open = true
@@ -170,3 +173,11 @@ func prepareMockDeclarePublisher(m *MockRawClient, publisherId uint8) {
 		gomock.AssignableToTypeOf(""),
 	)
 }
+
+func prepareMockDeletePublisher(m *MockRawClient, publisherId uint8) {
+	ctxType := reflect.TypeOf((*context.Context)(nil)).Elem()
+	m.EXPECT().DeletePublisher(
+		gomock.AssignableToTypeOf(ctxType),
+		gomock.Eq(publisherId),
+	)
+}