Fix stop pending producer on multiple mode requests

AlexxIT · May 14, 2024 · d9fb734 · d9fb734
1 parent a51156c
commit d9fb734
Showing 1 changed file with 5 additions and 0 deletions.
diff --git a/internal/streams/stream.go b/internal/streams/stream.go
@@ -88,6 +88,11 @@ func (s *Stream) RemoveProducer(prod core.Producer) {
 }
 
 func (s *Stream) stopProducers() {
+	if s.pending.Load() > 0 {
+		log.Trace().Msg("[streams] skip stop pending producer")
+		return
+	}
+
 	s.mu.Lock()
 producers:
 	for _, producer := range s.producers {