diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 5fbf35ebf..b9bcd98d3 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -12,6 +12,8 @@ import ( "fmt" "math" "net/http" + "sort" + "strings" "sync" "time" @@ -513,6 +515,7 @@ func (m *Manager) StartAll(ctx context.Context) error { dispatchCtx, cancel := context.WithCancel(ctx) m.dispatchTask = &asyncTask{cancel: cancel} + failedStarts := make(map[string]string) for name, channel := range m.channels { logger.InfoCF("channels", "Starting channel", map[string]any{ @@ -523,6 +526,8 @@ func (m *Manager) StartAll(ctx context.Context) error { "channel": name, "error": err.Error(), }) + fmt.Printf("Failed to start channel %s: %v\n", name, err) + failedStarts[name] = err.Error() continue } // Lazily create worker only after channel starts successfully @@ -532,6 +537,32 @@ func (m *Manager) StartAll(ctx context.Context) error { go m.runMediaWorker(dispatchCtx, name, w) } + if len(m.channels) > 0 && len(m.workers) == 0 { + if m.dispatchTask != nil { + m.dispatchTask.cancel() + m.dispatchTask = nil + } + + details := make([]string, 0, len(failedStarts)) + for name, reason := range failedStarts { + details = append(details, fmt.Sprintf("%s: %s", name, reason)) + } + sort.Strings(details) + if len(details) == 0 { + return fmt.Errorf("failed to start any enabled channels") + } + return fmt.Errorf("failed to start any enabled channels: %s", strings.Join(details, "; ")) + } + + if len(failedStarts) > 0 { + failedNames := make([]string, 0, len(failedStarts)) + for name := range failedStarts { + failedNames = append(failedNames, name) + } + sort.Strings(failedNames) + fmt.Printf("Warning: failed to start channels: %s\n", strings.Join(failedNames, ", ")) + } + // Start the dispatcher that reads from the bus and routes to workers go m.dispatchOutbound(dispatchCtx) go m.dispatchOutboundMedia(dispatchCtx)