Use reflect.Select to process dynamic list of channels correctly

Author: PhilippMatthes

internal/libvirt/libvirt.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"reflect"
 	"sync"
 	"time"
 
@@ -143,30 +144,49 @@ func (l *LibVirt) Close() error {
 func (l *LibVirt) runEventLoop(ctx context.Context) {
 	log := logger.FromContext(ctx, "libvirt", "event-loop")
 	for {
+		// The reflect.Select function works the same way as a
+		// regular select statement, but allows selecting over
+		// a dynamic set of channels.
+		var cases []reflect.SelectCase
+		var eventIds []libvirt.DomainEventID
 		for eventId, ch := range l.domEventChs {
-			select {
-			case <-ctx.Done():
-				return
-			case <-l.virt.Disconnected():
-				log.Error(errors.New("libvirt disconnected"), "waiting for reconnection")
-				time.Sleep(5 * time.Second)
-			case eventPayload, ok := <-ch:
-				if !ok {
-					err := errors.New("libvirt event channel closed")
-					log.Error(err, "eventId", eventId)
-					continue
-				}
-				handlers, exists := l.domEventChangeHandlers[eventId]
-				if !exists {
-					continue
-				}
-				for _, handler := range handlers {
-					// Process each handler sequentially.
-					handler(ctx, eventPayload)
-				}
-			default:
-				// No event available, continue
-			}
+			cases = append(cases, reflect.SelectCase{
+				Dir:  reflect.SelectRecv,
+				Chan: reflect.ValueOf(ch),
+			})
+			eventIds = append(eventIds, eventId)
+		}
+
+		cases = append(cases, reflect.SelectCase{
+			Dir:  reflect.SelectRecv,
+			Chan: reflect.ValueOf(ctx.Done()),
+		})
+		caseCtxDone := len(cases) - 1
+
+		chosen, value, ok := reflect.Select(cases)
+		if !ok {
+			// This should never happen. If it does, give the
+			// service a chance to restart and reconnect.
+			panic("libvirt connection closed")
+		}
+		if chosen == caseCtxDone {
+			log.Info("shutting down libvirt event loop")
+			return
+		}
+		if chosen >= len(eventIds) {
+			msg := "no handler for selected channel"
+			log.Error(errors.New("invalid event channel selected"), msg)
+			continue
+		}
+
+		// Distribute the event to all registered handlers.
+		eventId := eventIds[chosen] // safe as chosen < len(eventIds)
+		handlers, exists := l.domEventChangeHandlers[eventId]
+		if !exists {
+			continue
+		}
+		for _, handler := range handlers {
+			handler(ctx, value.Interface())
 		}
 	}
 }

internal/libvirt/libvirt_test.go

@@ -1025,10 +1025,13 @@ func TestWatchDomainChanges_OverwriteHandler(t *testing.T) {
 }
 
 func TestRunEventLoop_ProcessesEvents(t *testing.T) {
-	// Create a channel for events
-	eventCh := make(chan any, 1)
+	// Create a buffered channel for events that won't be closed during the test
+	eventChInternal := make(chan any, 10)
+
+	// Wrap it in a read-only channel to prevent accidental closure
+	var eventCh <-chan any = eventChInternal
 
-	mockConn := &mockLibvirtConnection{}
+	mockConn := newMockLibvirtConnection()
 
 	l := &LibVirt{
 		virt: &mockConn.Libvirt,
@@ -1048,16 +1051,12 @@ func TestRunEventLoop_ProcessesEvents(t *testing.T) {
 
 	l.WatchDomainChanges(libvirt.DomainEventIDLifecycle, "test-handler", handler)
 
-	// Create a context with timeout
-	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
-	defer cancel()
-
 	// Start the event loop in a goroutine
-	go l.runEventLoop(ctx)
+	go l.runEventLoop(t.Context())
 
 	// Send an event
 	testPayload := "test-event-payload"
-	eventCh <- testPayload
+	eventChInternal <- testPayload
 
 	// Give some time for the event to be processed
 	time.Sleep(50 * time.Millisecond)
@@ -1078,6 +1077,12 @@ type mockLibvirtConnection struct {
 	disconnectedCh chan struct{}
 }
 
+func newMockLibvirtConnection() *mockLibvirtConnection {
+	return &mockLibvirtConnection{
+		disconnectedCh: make(chan struct{}),
+	}
+}
+
 func (m *mockLibvirtConnection) Disconnected() <-chan struct{} {
 	return m.disconnectedCh
 }