From 241c81dc566c558cefd950ff6434a746503d5b2b Mon Sep 17 00:00:00 2001 From: aler9 <46489434+aler9@users.noreply.github.com> Date: Tue, 27 Dec 2022 12:00:00 +0000 Subject: [PATCH] hls source: download first playlist once --- internal/hls/client_downloader_stream.go | 23 ++-- internal/hls/client_test.go | 159 ++++++++++++----------- 2 files changed, 100 insertions(+), 82 deletions(-) diff --git a/internal/hls/client_downloader_stream.go b/internal/hls/client_downloader_stream.go index d7e15e8c..aa506f15 100644 --- a/internal/hls/client_downloader_stream.go +++ b/internal/hls/client_downloader_stream.go @@ -36,7 +36,7 @@ func findSegmentWithInvPosition(segments []*gm3u8.MediaSegment, pos int) *gm3u8. func findSegmentWithID(seqNo uint64, segments []*gm3u8.MediaSegment, id uint64) (*gm3u8.MediaSegment, int) { index := int(int64(id) - int64(seqNo)) - if (index) >= len(segments) { + if index >= len(segments) { return nil, 0 } @@ -139,13 +139,23 @@ func (d *clientDownloaderStream) run(ctx context.Context) error { d.rp.add(proc) } + err := d.fillSegmentQueue(ctx, initialPlaylist, segmentQueue) + if err != nil { + return err + } + for { ok := segmentQueue.waitUntilSizeIsBelow(ctx, 1) if !ok { return fmt.Errorf("terminated") } - err := d.fillSegmentQueue(ctx, segmentQueue) + pl, err := d.downloadPlaylist(ctx) + if err != nil { + return err + } + + err = d.fillSegmentQueue(ctx, pl, segmentQueue) if err != nil { return err } @@ -204,12 +214,9 @@ func (d *clientDownloaderStream) downloadSegment(ctx context.Context, return byts, nil } -func (d *clientDownloaderStream) fillSegmentQueue(ctx context.Context, segmentQueue *clientSegmentQueue) error { - pl, err := d.downloadPlaylist(ctx) - if err != nil { - return err - } - +func (d *clientDownloaderStream) fillSegmentQueue(ctx context.Context, + pl *m3u8.MediaPlaylist, segmentQueue *clientSegmentQueue, +) error { pl.Segments = pl.Segments[:segmentsLen(pl.Segments)] var seg *gm3u8.MediaSegment diff --git a/internal/hls/client_test.go b/internal/hls/client_test.go index cffa595b..be42c7d8 100644 --- a/internal/hls/client_test.go +++ b/internal/hls/client_test.go @@ -93,82 +93,53 @@ func writeTempFile(byts []byte) (string, error) { return tmpf.Name(), nil } +func mpegtsSegment(w io.Writer) { + mux := astits.NewMuxer(context.Background(), w) + mux.AddElementaryStream(astits.PMTElementaryStream{ + ElementaryPID: 256, + StreamType: astits.StreamTypeH264Video, + }) + mux.SetPCRPID(256) + mux.WriteTables() + + enc, _ := h264.AnnexBMarshal([][]byte{ + {7, 1, 2, 3}, // SPS + {8}, // PPS + {5}, // IDR + }) + + mux.WriteData(&astits.MuxerData{ + PID: 256, + PES: &astits.PESData{ + Header: &astits.PESHeader{ + OptionalHeader: &astits.PESOptionalHeader{ + MarkerBits: 2, + PTSDTSIndicator: astits.PTSDTSIndicatorBothPresent, + PTS: &astits.ClockReference{Base: 90000}, // +1 sec + DTS: &astits.ClockReference{Base: 0x1FFFFFFFF - 90000 + 1}, // -1 sec + }, + StreamID: 224, // = video + }, + Data: enc, + }, + }) +} + type testHLSServer struct { s *http.Server } -func newTestHLSServer(ca string) (*testHLSServer, error) { +func newTestHLSServer(router http.Handler, isTLS bool) (*testHLSServer, error) { ln, err := net.Listen("tcp", "localhost:5780") if err != nil { return nil, err } - ts := &testHLSServer{} - - gin.SetMode(gin.ReleaseMode) - router := gin.New() - - segment := "segment.ts" - if ca == "segment with query" { - segment = "segment.ts?key=val" + s := &testHLSServer{ + s: &http.Server{Handler: router}, } - router.GET("/stream.m3u8", func(ctx *gin.Context) { - cnt := `#EXTM3U -#EXT-X-VERSION:3 -#EXT-X-ALLOW-CACHE:NO -#EXT-X-TARGETDURATION:2 -#EXT-X-MEDIA-SEQUENCE:0 -#EXTINF:2, -` + segment + ` -#EXT-X-ENDLIST -` - - ctx.Writer.Header().Set("Content-Type", `application/x-mpegURL`) - io.Copy(ctx.Writer, bytes.NewReader([]byte(cnt))) - }) - - router.GET("/segment.ts", func(ctx *gin.Context) { - if ca == "segment with query" && ctx.Query("key") != "val" { - return - } - - ctx.Writer.Header().Set("Content-Type", `video/MP2T`) - - mux := astits.NewMuxer(context.Background(), ctx.Writer) - mux.AddElementaryStream(astits.PMTElementaryStream{ - ElementaryPID: 256, - StreamType: astits.StreamTypeH264Video, - }) - mux.SetPCRPID(256) - mux.WriteTables() - - enc, _ := h264.AnnexBMarshal([][]byte{ - {7, 1, 2, 3}, // SPS - {8}, // PPS - {5}, // IDR - }) - - mux.WriteData(&astits.MuxerData{ - PID: 256, - PES: &astits.PESData{ - Header: &astits.PESHeader{ - OptionalHeader: &astits.PESOptionalHeader{ - MarkerBits: 2, - PTSDTSIndicator: astits.PTSDTSIndicatorBothPresent, - PTS: &astits.ClockReference{Base: 90000}, // +1 sec - DTS: &astits.ClockReference{Base: 0x1FFFFFFFF - 90000 + 1}, // -1 sec - }, - StreamID: 224, // = video - }, - Data: enc, - }, - }) - }) - - ts.s = &http.Server{Handler: router} - - if ca == "tls" { + if isTLS { go func() { serverCertFpath, err := writeTempFile(serverCert) if err != nil { @@ -182,17 +153,17 @@ func newTestHLSServer(ca string) (*testHLSServer, error) { } defer os.Remove(serverKeyFpath) - ts.s.ServeTLS(ln, serverCertFpath, serverKeyFpath) + s.s.ServeTLS(ln, serverCertFpath, serverKeyFpath) }() } else { - go ts.s.Serve(ln) + go s.s.Serve(ln) } - return ts, nil + return s, nil } -func (ts *testHLSServer) close() { - ts.s.Shutdown(context.Background()) +func (s *testHLSServer) close() { + s.s.Shutdown(context.Background()) } func TestClient(t *testing.T) { @@ -202,9 +173,44 @@ func TestClient(t *testing.T) { "segment with query", } { t.Run(ca, func(t *testing.T) { - ts, err := newTestHLSServer(ca) + gin.SetMode(gin.ReleaseMode) + router := gin.New() + + segment := "segment.ts" + if ca == "segment with query" { + segment = "segment.ts?key=val" + } + sent := false + + router.GET("/stream.m3u8", func(ctx *gin.Context) { + if sent { + return + } + sent = true + + ctx.Writer.Header().Set("Content-Type", `application/x-mpegURL`) + io.Copy(ctx.Writer, bytes.NewReader([]byte(`#EXTM3U + #EXT-X-VERSION:3 + #EXT-X-ALLOW-CACHE:NO + #EXT-X-TARGETDURATION:2 + #EXT-X-MEDIA-SEQUENCE:0 + #EXTINF:2, + `+segment+` + #EXT-X-ENDLIST + `))) + }) + + router.GET("/segment.ts", func(ctx *gin.Context) { + if ca == "segment with query" { + require.Equal(t, "val", ctx.Query("key")) + } + ctx.Writer.Header().Set("Content-Type", `video/MP2T`) + mpegtsSegment(ctx.Writer) + }) + + s, err := newTestHLSServer(router, ca == "tls") require.NoError(t, err) - defer ts.close() + defer s.close() packetRecv := make(chan struct{}) @@ -216,7 +222,12 @@ func TestClient(t *testing.T) { c, err := NewClient( prefix+"://localhost:5780/stream.m3u8", "33949E05FFFB5FF3E8AA16F8213A6251B4D9363804BA53233C4DA9A46D6F2739", - func(*format.H264, *format.MPEG4Audio) error { + func(videoTrack *format.H264, audioTrack *format.MPEG4Audio) error { + require.Equal(t, &format.H264{ + PayloadTyp: 96, + PacketizationMode: 1, + }, videoTrack) + require.Equal(t, (*format.MPEG4Audio)(nil), audioTrack) return nil }, func(pts time.Duration, nalus [][]byte) { @@ -237,7 +248,7 @@ func TestClient(t *testing.T) { <-packetRecv c.Close() - c.Wait() + <-c.Wait() }) } }