Method broadcastLoop

internal/core/local_service.go:146–195

()

}

func (s *LocalDownloadService) broadcastLoop() {
	for msg := range s.InputCh {
		s.listenerMu.Lock()
		listenersCopy := make([]chan any, len(s.listeners))
		copy(listenersCopy, s.listeners)
		s.listenerMu.Unlock()

		for _, ch := range listenersCopy {
			// Check message type
			isProgress := false
			switch msg.(type) {
			case events.ProgressMsg:
				isProgress = true
			case events.BatchProgressMsg:
				isProgress = true
			}

			func() {
				defer func() { _ = recover() }()
				if isProgress {
					// Non-blocking send for progress updates
					select {
					case ch <- msg:
					default:
						// Drop progress message if channel is full
					}
				} else {
					// Blocking send with timeout for critical state changes
					// We don't want to drop these, but we also don't want to block forever if a client is dead
					select {
					case ch <- msg:
					case <-time.After(1 * time.Second):
						utils.Debug("Dropped critical event due to slow client")
					}
				}
			}()
		}
	}
	// Close all listeners when input closes
	s.listenerMu.Lock()
	for _, ch := range s.listeners {
		close(ch)
	}
	s.listeners = nil
	s.listenerMu.Unlock()

	if s.reportTicker != nil {
		s.reportTicker.Stop()
	}
}

func (s *LocalDownloadService) reportProgressLoop() {
	lastSpeeds := make(map[string]float64)