fix(queue): prevent Broadcast from blocking on closed/empty channels

[Hlgxz]

🐛 Bug Fix: Prevent Queue.Broadcast from Blocking on Closed/Empty Channels

Problem Description

The Queue.Broadcast method has a potential goroutine leak issue when consumers have exited but their event channels are still in the eventChannels list. This can cause the Broadcast goroutine to block permanently when trying to send messages to channels with no receivers.

Root Cause Analysis

1. Channel Lifecycle Issue:

   • When startConsumers creates consumers, it creates unbuffered eventChan channels and adds them to eventChannels list
   • When consumers exit (after q.channel closes), the consume function returns, but the eventChan is never removed from eventChannels
   • The channels are never closed, leaving stale references in the list

2. Blocking Behavior:

   • Broadcast sends messages to all channels in eventChannels using blocking send: channel <- message
   • If a consumer has exited, there's no receiver for its eventChan
   • Since eventChan is an unbuffered channel, the send operation blocks forever
   • This causes goroutine leaks as the Broadcast goroutine never completes

Reproduction Steps

// Scenario that can trigger the issue:
q := NewQueue(producerFactory, consumerFactory)
q.Start()  // Consumers start and eventChannels are populated

// ... later, producers finish and q.channel closes
// Consumers exit, but eventChannels list is not cleaned up

// If Broadcast is called after consumers exit:
q.Broadcast("message")  // Goroutine leaks - blocks forever on stale channels

Solution

Use a non-blocking select statement with a default case to skip channels that cannot receive messages:

for _, channel := range q.eventChannels {
    select {
    case channel <- message:
        // Successfully sent
    default:
        // Channel is full or has no receiver, skip to avoid blocking
        // This can happen if the consumer has exited but the channel
        // is still in the eventChannels list
    }
}

This approach:

• Prevents goroutine leaks by avoiding permanent blocking

Testing

• All existing tests pass, including TestQueue_Broadcast
• The fix handles both normal operation (consumers active) and edge cases (consumers exited)

Related Issues

This fix addresses a potential resource leak that could occur in long-running applications where Queue instances are reused or where Broadcast is called after consumers have exited.

[Hlgxz]

@kevwan Could you please take a look at this fix? All existing tests are passing, and we’d appreciate getting this merged soon. The change uses a non‑blocking send to avoid goroutine leaks when consumers have already exited. Thanks!