tuner/backend/stream/hub.go

// Package stream implements a single-upstream, many-client fan-out hub for
// raw MPEG-TS streams. Only one connection is ever opened to the upstream IPTV
// source at a time; all connected browsers receive a copy of the same bytes.
package stream

import (
	"context"
	"io"
	"log"
	"net/http"
	"sync"
	"time"
)

// chunkSize is the read buffer size for pulling from the upstream source.
const chunkSize = 64 * 1024

// subscriberBuffer is how many chunks may queue per client before it is
// considered too slow and dropped (prevents one slow client stalling others).
const subscriberBuffer = 256

// Subscriber receives copies of upstream TS chunks.
type subscriber struct {
	ch     chan []byte
	closed chan struct{}
}

// Hub manages a single shared upstream connection and fans its bytes out to
// any number of subscribers. Switching channels tears down the current
// upstream and starts a new one.
type Hub struct {
	mu sync.Mutex

	streamURL   string
	cancel      context.CancelFunc
	subscribers map[*subscriber]struct{}
	running     bool

	// gotData is closed once the first upstream bytes arrive after a switch;
	// used to detect upstream-down conditions.
	gotData   bool
	startedAt time.Time

	client *http.Client
}

// NewHub creates an empty hub.
func NewHub() *Hub {
	return &Hub{
		subscribers: make(map[*subscriber]struct{}),
		client: &http.Client{
			// No overall timeout: this is a long-lived stream. Per-attempt
			// dial/response-header timeouts are set on the transport.
			Transport: &http.Transport{
				ResponseHeaderTimeout: 15 * time.Second,
				IdleConnTimeout:       30 * time.Second,
			},
		},
	}
}

// SetChannel switches the shared upstream to streamURL. If the same URL is
// already active, it is a no-op. An empty URL stops streaming.
func (h *Hub) SetChannel(streamURL string) {
	h.mu.Lock()
	defer h.mu.Unlock()

	if streamURL == h.streamURL && h.running {
		return
	}

	// Tear down existing upstream
	h.stopLocked()

	h.streamURL = streamURL
	if streamURL == "" {
		return
	}

	ctx, cancel := context.WithCancel(context.Background())
	h.cancel = cancel
	h.running = true
	h.gotData = false
	h.startedAt = time.Now()

	go h.run(ctx, streamURL)
	log.Printf("[hub] switched upstream to %s", streamURL)
}

// Stop closes the upstream connection and clears the active channel.
func (h *Hub) Stop() {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.streamURL = ""
	h.stopLocked()
}

func (h *Hub) stopLocked() {
	if h.cancel != nil {
		h.cancel()
		h.cancel = nil
	}
	h.running = false
}

// GotData reports whether upstream bytes have been received since the last
// channel switch, plus how long ago the switch occurred.
func (h *Hub) GotData() (bool, time.Duration) {
	h.mu.Lock()
	defer h.mu.Unlock()
	if !h.running {
		return false, 0
	}
	return h.gotData, time.Since(h.startedAt)
}

// Subscribe registers a new client. Returns a subscriber and an unsubscribe
// function the caller must invoke when done.
func (h *Hub) Subscribe() (*subscriber, func()) {
	sub := &subscriber{
		ch:     make(chan []byte, subscriberBuffer),
		closed: make(chan struct{}),
	}

	h.mu.Lock()
	h.subscribers[sub] = struct{}{}
	count := len(h.subscribers)
	h.mu.Unlock()

	log.Printf("[hub] client subscribed (%d total)", count)

	var once sync.Once
	unsub := func() {
		once.Do(func() {
			h.mu.Lock()
			delete(h.subscribers, sub)
			remaining := len(h.subscribers)
			h.mu.Unlock()
			close(sub.closed)
			log.Printf("[hub] client unsubscribed (%d remaining)", remaining)
		})
	}
	return sub, unsub
}

// Chan exposes the subscriber's chunk channel.
func (s *subscriber) Chan() <-chan []byte { return s.ch }

// Closed signals the subscriber has been removed by the hub.
func (s *subscriber) Done() <-chan struct{} { return s.closed }

// broadcast sends a chunk to all subscribers, dropping any that are too slow.
func (h *Hub) broadcast(chunk []byte) {
	h.mu.Lock()
	if !h.gotData {
		h.gotData = true
	}
	subs := make([]*subscriber, 0, len(h.subscribers))
	for s := range h.subscribers {
		subs = append(subs, s)
	}
	h.mu.Unlock()

	for _, s := range subs {
		select {
		case s.ch <- chunk:
		default:
			// Slow client: drop this chunk for them rather than blocking everyone.
		}
	}
}

// run maintains the upstream connection, reconnecting on failure until the
// context is cancelled (channel switch/stop).
func (h *Hub) run(ctx context.Context, streamURL string) {
	backoff := time.Second
	for {
		if ctx.Err() != nil {
			return
		}
		err := h.pull(ctx, streamURL)
		if ctx.Err() != nil {
			return
		}
		if err != nil {
			log.Printf("[hub] upstream error (%s), retrying in %s: %v", streamURL, backoff, err)
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(backoff):
		}
		if backoff < 10*time.Second {
			backoff *= 2
		}
	}
}

// pull opens a single upstream connection and copies bytes to subscribers
// until the stream ends or the context is cancelled.
func (h *Hub) pull(ctx context.Context, streamURL string) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, streamURL, nil)
	if err != nil {
		return err
	}
	req.Header.Set("User-Agent", "VLC/3.0.20 LibVLC/3.0.20")

	resp, err := h.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return &httpError{status: resp.StatusCode}
	}

	buf := make([]byte, chunkSize)
	for {
		n, err := resp.Body.Read(buf)
		if n > 0 {
			chunk := make([]byte, n)
			copy(chunk, buf[:n])
			h.broadcast(chunk)
		}
		if err != nil {
			if err == io.EOF {
				return nil
			}
			return err
		}
		if ctx.Err() != nil {
			return ctx.Err()
		}
	}
}

type httpError struct{ status int }

func (e *httpError) Error() string {
	return "upstream returned HTTP status " + http.StatusText(e.status)
}