feat(v2): add reconnecting FFmpeg stream source

v2/internal/source/stream_test.go

@@ -0,0 +1,92 @@
+package source
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestPublishLatestDropsStaleFrame(t *testing.T) {
+	frames := make(chan Frame, 1)
+	publishLatest(frames, Frame{Sequence: 1})
+	publishLatest(frames, Frame{Sequence: 2})
+
+	if got := <-frames; got.Sequence != 2 {
+		t.Fatalf("sequence = %d, want 2", got.Sequence)
+	}
+}
+
+func TestRetryDelayIsBounded(t *testing.T) {
+	if got := retryDelay(0); got != time.Second {
+		t.Fatalf("first retry delay = %s", got)
+	}
+	if got := retryDelay(3); got != 8*time.Second {
+		t.Fatalf("fourth retry delay = %s", got)
+	}
+	if got := retryDelay(99); got != 8*time.Second {
+		t.Fatalf("bounded retry delay = %s", got)
+	}
+}
+
+func TestStreamRetriesAndPublishesLatestCompletedFrame(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var decodeCalls int32
+	decodedLatest := make(chan struct{})
+	stream := Start(ctx, Config{SourceURL: "rtsp://operator:secret@example/101"}, Dependencies{
+		Probe: func(context.Context, Config) (Metadata, error) {
+			return Metadata{Width: 2, Height: 1, FPS: 30}, nil
+		},
+		Decode: func(ctx context.Context, _ Config, _ Metadata, publish func([]byte)) error {
+			call := atomic.AddInt32(&decodeCalls, 1)
+			if call == 1 {
+				publish([]byte{1, 1, 1, 1, 1, 1})
+				return errors.New("temporary decoder failure")
+			}
+			publish([]byte{2, 2, 2, 2, 2, 2})
+			close(decodedLatest)
+			cancel()
+			return ctx.Err()
+		},
+		Wait: func(ctx context.Context, _ time.Duration) error { return ctx.Err() },
+	})
+
+	select {
+	case <-decodedLatest:
+	case <-time.After(time.Second):
+		t.Fatal("decoder did not publish the second frame")
+	}
+	select {
+	case got := <-stream.Frames():
+		if got.Sequence != 2 || got.BGR[0] != 2 {
+			t.Fatalf("frame = %#v", got)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("did not receive latest decoded frame")
+	}
+	<-stream.Done()
+}
+
+func TestStreamNeverPublishesDecoderErrorDetails(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	stream := Start(ctx, Config{SourceURL: "rtsp://operator:secret@example/101"}, Dependencies{
+		Probe: func(context.Context, Config) (Metadata, error) {
+			return Metadata{}, errors.New("ffprobe rtsp://operator:secret@example/101 failed")
+		},
+		Decode: func(context.Context, Config, Metadata, func([]byte)) error { return nil },
+		Wait: func(ctx context.Context, _ time.Duration) error {
+			cancel()
+			return ctx.Err()
+		},
+	})
+
+	for update := range stream.Statuses() {
+		if strings.Contains(update.Message, "secret") || strings.Contains(update.Message, "rtsp://") {
+			t.Fatalf("status leaked source details: %#v", update)
+		}
+	}
+}