fm-rds-tx/internal/ingest/adapters/aoip/source_test.go

package aoip

import (
	"context"
	"testing"
	"time"

	"aoiprxkit"
	"github.com/jan/fm-rds-tx/internal/ingest"
)

type stubReceiver struct {
	onStart func()
	onStop  func()
	stats   aoiprxkit.Stats
}

func (r *stubReceiver) Start(context.Context) error {
	if r.onStart != nil {
		r.onStart()
	}
	return nil
}

func (r *stubReceiver) Stop() error {
	if r.onStop != nil {
		r.onStop()
	}
	return nil
}

func (r *stubReceiver) Stats() aoiprxkit.Stats {
	return r.stats
}

func TestSourceEmitsChunksAndMapsStats(t *testing.T) {
	var handler aoiprxkit.FrameHandler
	rx := &stubReceiver{
		stats: aoiprxkit.Stats{
			PacketsGapLoss:  1,
			PacketsLateDrop: 2,
			JitterReorders:  1,
		},
	}
	src := New("aes67-test", aoiprxkit.Config{
		MulticastGroup:     "239.10.20.30",
		Port:               5004,
		PayloadType:        97,
		SampleRateHz:       48000,
		Channels:           2,
		Encoding:           "L24",
		PacketTime:         time.Millisecond,
		JitterDepthPackets: 6,
	}, WithReceiverFactory(func(_ aoiprxkit.Config, onFrame aoiprxkit.FrameHandler) (ReceiverClient, error) {
		handler = onFrame
		return rx, nil
	}))

	if err := src.Start(context.Background()); err != nil {
		t.Fatalf("start: %v", err)
	}
	defer src.Stop()

	handler(aoiprxkit.PCMFrame{
		SequenceNumber: 100,
		SampleRateHz:   48000,
		Channels:       2,
		Samples:        []int32{1, -1, 2, -2},
		ReceivedAt:     time.Now(),
	})
	handler(aoiprxkit.PCMFrame{
		SequenceNumber: 102,
		SampleRateHz:   48000,
		Channels:       2,
		Samples:        []int32{3, -3, 4, -4},
		ReceivedAt:     time.Now(),
	})

	chunk1 := readChunk(t, src.Chunks())
	if chunk1.Discontinuity {
		t.Fatalf("first chunk should not be discontinuity")
	}
	chunk2 := readChunk(t, src.Chunks())
	if !chunk2.Discontinuity {
		t.Fatalf("second chunk should be discontinuity on sequence gap")
	}

	stats := src.Stats()
	if stats.State != "running" {
		t.Fatalf("state=%q want running", stats.State)
	}
	if !stats.Connected {
		t.Fatalf("connected=false want true")
	}
	if stats.ChunksIn != 2 {
		t.Fatalf("chunksIn=%d want 2", stats.ChunksIn)
	}
	if stats.SamplesIn != 8 {
		t.Fatalf("samplesIn=%d want 8", stats.SamplesIn)
	}
	if stats.TransportLoss != 1 {
		t.Fatalf("transportLoss=%d want 1", stats.TransportLoss)
	}
	if stats.Reorders != 1 {
		t.Fatalf("reorders=%d want 1", stats.Reorders)
	}
	if stats.Underruns != 2 {
		t.Fatalf("underruns=%d want 2", stats.Underruns)
	}
	if stats.JitterDepth != 6 {
		t.Fatalf("jitterDepth=%d want 6", stats.JitterDepth)
	}
}

func TestSourceDescriptorSupportsDetailOverride(t *testing.T) {
	src := New("aes67-test", aoiprxkit.Config{
		MulticastGroup: "239.10.20.30",
		Port:           5004,
		SampleRateHz:   48000,
		Channels:       2,
	}, WithDetail("rtp://239.10.20.30:5004 (SAP s=AES67-MAIN)"), WithOrigin(ingest.SourceOrigin{
		Kind:       "sap-discovery",
		StreamName: "AES67-MAIN",
		Endpoint:   "rtp://239.10.20.30:5004",
	}))

	desc := src.Descriptor()
	if desc.Detail != "rtp://239.10.20.30:5004 (SAP s=AES67-MAIN)" {
		t.Fatalf("detail=%q", desc.Detail)
	}
	if desc.Origin == nil {
		t.Fatalf("expected descriptor origin")
	}
	if desc.Origin.Kind != "sap-discovery" {
		t.Fatalf("origin kind=%q", desc.Origin.Kind)
	}
	if desc.Origin.StreamName != "AES67-MAIN" {
		t.Fatalf("origin streamName=%q", desc.Origin.StreamName)
	}
}

func readChunk(t *testing.T, ch <-chan ingest.PCMChunk) ingest.PCMChunk {
	t.Helper()
	select {
	case chunk, ok := <-ch:
		if !ok {
			t.Fatal("chunk channel closed")
		}
		return chunk
	case <-time.After(500 * time.Millisecond):
		t.Fatal("timeout waiting for chunk")
		return ingest.PCMChunk{}
	}
}