package srt import ( "bytes" "context" "io" "testing" "time" "aoiprxkit" "github.com/jan/fm-rds-tx/internal/ingest" ) type readCloser struct{ io.Reader } func (r readCloser) Close() error { return nil } func TestSourceEmitsChunksFromSRTFrames(t *testing.T) { var stream bytes.Buffer if err := aoiprxkit.WritePCM32Packet(&stream, 2, 48000, 2, 10, 100, []int32{1, 2, 3, 4}); err != nil { t.Fatalf("write packet 1: %v", err) } if err := aoiprxkit.WritePCM32Packet(&stream, 2, 48000, 2, 12, 200, []int32{5, 6, 7, 8}); err != nil { t.Fatalf("write packet 2: %v", err) } src := New("srt-test", aoiprxkit.SRTConfig{ URL: "srt://127.0.0.1:9000?mode=listener", Mode: "listener", SampleRateHz: 48000, Channels: 2, }, WithConnOpener(func(ctx context.Context, cfg aoiprxkit.SRTConfig) (io.ReadCloser, error) { _ = ctx _ = cfg return readCloser{Reader: bytes.NewReader(stream.Bytes())}, nil })) desc := src.Descriptor() if desc.Origin == nil { t.Fatalf("expected descriptor origin") } if desc.Origin.Kind != "url" { t.Fatalf("origin kind=%q want url", desc.Origin.Kind) } if desc.Origin.Endpoint != "srt://127.0.0.1:9000" { t.Fatalf("origin endpoint=%q", desc.Origin.Endpoint) } if desc.Origin.Mode != "listener" { t.Fatalf("origin mode=%q want listener", desc.Origin.Mode) } if err := src.Start(context.Background()); err != nil { t.Fatalf("start: %v", err) } defer src.Stop() chunk1 := readChunk(t, src.Chunks()) if chunk1.SourceID != "srt-test" { t.Fatalf("source id=%q want srt-test", chunk1.SourceID) } if chunk1.Channels != 2 || chunk1.SampleRateHz != 48000 { t.Fatalf("shape=%d/%d", chunk1.Channels, chunk1.SampleRateHz) } if chunk1.Discontinuity { t.Fatalf("first chunk should not be discontinuity") } assertSamples(t, chunk1.Samples, []int32{1, 2, 3, 4}) chunk2 := readChunk(t, src.Chunks()) if !chunk2.Discontinuity { t.Fatalf("second chunk should be marked discontinuity on seq gap") } assertSamples(t, chunk2.Samples, []int32{5, 6, 7, 8}) 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.Discontinuities < 1 { t.Fatalf("discontinuities=%d want >=1", stats.Discontinuities) } if stats.LastChunkAt.IsZero() { t.Fatalf("lastChunkAt should be set") } } 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{} } } func assertSamples(t *testing.T, got, want []int32) { t.Helper() if len(got) != len(want) { t.Fatalf("sample len=%d want %d", len(got), len(want)) } for i := range want { if got[i] != want[i] { t.Fatalf("sample[%d]=%d want %d", i, got[i], want[i]) } } }