diff --git a/pkg/api/api.go b/pkg/api/api.go index 23b466d61..80ce35982 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -87,8 +87,6 @@ type StreamplaceAPI struct { // override tls port for http redirect server if we're using systemd file descriptors HTTPRedirectTLSPort *int - sessions map[string]map[string]time.Time - sessionsLock sync.RWMutex rtmpSessions map[string]*media.RTMPSession rtmpSessionsLock sync.Mutex @@ -124,8 +122,6 @@ func MakeStreamplaceAPI(cli *config.CLI, mod model.Model, statefulDB *statedb.St limiters: make(map[string]*rate.Limiter), SignerCache: make(map[string]media.MediaSigner), op: op, - sessions: make(map[string]map[string]time.Time), - sessionsLock: sync.RWMutex{}, rtmpSessions: make(map[string]*media.RTMPSession), rtmpSessionsLock: sync.Mutex{}, LocalDB: ldb, @@ -202,7 +198,6 @@ func (a *StreamplaceAPI) Handler(ctx context.Context) (http.Handler, error) { addHandle(apiRouter, "DELETE", "/api/webrtc/:stream", a.MistProxyHandler(ctx, "/webrtc/%s")) addFunc(apiRouter, "POST", "/api/segment", a.HandleSegment(ctx)) addFunc(apiRouter, "GET", "/api/healthz", a.HandleHealthz(ctx)) - addHandle(apiRouter, "GET", "/api/playback/:user/hls/*file", a.HandleHLSPlayback(ctx)) // they're jpegs now addHandle(apiRouter, "GET", "/api/playback/:user/stream.jpg", a.HandleThumbnailPlayback(ctx)) // this one is actually a jpeg (used previously and shouldn't remove for historical reasons) diff --git a/pkg/api/playback.go b/pkg/api/playback.go index b5346adae..055cee5af 100644 --- a/pkg/api/playback.go +++ b/pkg/api/playback.go @@ -1,7 +1,6 @@ package api import ( - "bytes" "context" "fmt" "io" @@ -14,7 +13,6 @@ import ( "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/errors" "stream.place/streamplace/pkg/log" - "stream.place/streamplace/pkg/spmetrics" ) func (a *StreamplaceAPI) NormalizeUser(ctx context.Context, user string) (string, error) { @@ -176,88 +174,6 @@ func NoCache(h httprouter.Handle) httprouter.Handle { } } -const SessionExpireTime = 30 * time.Second - -func (a *StreamplaceAPI) SessionSeen(ctx context.Context, user string, session string) { - now := time.Now() - go func() { - a.sessionsLock.Lock() - defer a.sessionsLock.Unlock() - if _, ok := a.sessions[user]; !ok { - a.sessions[user] = map[string]time.Time{} - } - if _, ok := a.sessions[user][session]; !ok { - log.Warn(ctx, "ViewerInc", "user", user, "session", session) - spmetrics.ViewerInc(user, "hls") - a.Bus.IncrementViewerCount(user, "local") - } - a.sessions[user][session] = now - }() -} - -func (a *StreamplaceAPI) ExpireSessions(ctx context.Context) error { - for { - select { - case <-ctx.Done(): - return nil - case <-time.After(5 * time.Second): - a.sessionsLock.Lock() - for user, sessions := range a.sessions { - for session, seen := range sessions { - if time.Since(seen) > SessionExpireTime { - delete(sessions, session) - spmetrics.ViewerDec(user, "hls") - a.Bus.DecrementViewerCount(user, "local") - } - } - } - a.sessionsLock.Unlock() - } - } -} - -func (a *StreamplaceAPI) HandleHLSPlayback(ctx context.Context) httprouter.Handle { - return NoCache(func(w http.ResponseWriter, r *http.Request, p httprouter.Params) { - user := p.ByName("user") - if user == "" { - errors.WriteHTTPBadRequest(w, "user required", nil) - return - } - user, err := a.NormalizeUser(ctx, user) - if err != nil { - errors.WriteHTTPBadRequest(w, "invalid user", err) - return - } - file := p.ByName("file") - if file == "" { - errors.WriteHTTPBadRequest(w, "file required", nil) - return - } - m3u8, err := a.Director.GetM3U8(ctx, user) - if err != nil { - errors.WriteHTTPNotFound(w, "could not get m3u8", err) - return - } - session := r.URL.Query().Get("session") - rendition := r.URL.Query().Get("rendition") - buf, err := m3u8.GetFile(file, session, rendition) - if err != nil { - errors.WriteHTTPNotFound(w, "segment not found", err) - return - } - - if strings.HasSuffix(file, ".m3u8") { - w.Header().Set("Content-Type", "application/vnd.apple.mpegurl") - } else { - if session != "" { - a.SessionSeen(ctx, user, session) - } - w.Header().Set("Content-Type", "video/mp2t") - } - - http.ServeContent(w, r, file, time.Now(), bytes.NewReader(buf)) - }) -} // thumbnailMaxAge is how stale a thumbnail may be before we treat the user as // offline and stop serving it. It must comfortably exceed thumbnailInterval (the diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index 0d3915bda..e4393ee91 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -527,10 +527,6 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu }) } - group.Go(func() error { - return a.ExpireSessions(ctx) - }) - group.Go(func() error { return storage.StartSegmentCleaner(ctx, ldb, cli) }) diff --git a/pkg/director/director.go b/pkg/director/director.go index 81a05344a..1ff38d4ef 100644 --- a/pkg/director/director.go +++ b/pkg/director/director.go @@ -2,7 +2,6 @@ package director import ( "context" - "fmt" "sync" "github.com/streamplace/oatproxy/pkg/oatproxy" @@ -69,7 +68,6 @@ func (d *Director) Start(ctx context.Context) error { ss, ok := d.streamSessions[not.Segment.RepoDID] if !ok { ss = &StreamSession{ - hls: nil, lp: nil, repoDID: not.Segment.RepoDID, mm: d.mm, @@ -112,12 +110,3 @@ func (d *Director) Start(ctx context.Context) error { } } -func (d *Director) GetM3U8(ctx context.Context, repoDID string) (*media.M3U8, error) { - d.streamSessionsMu.Lock() - defer d.streamSessionsMu.Unlock() - ss, ok := d.streamSessions[repoDID] - if !ok { - return nil, fmt.Errorf("stream session not found") - } - return ss.hls, nil -} diff --git a/pkg/director/stream_session.go b/pkg/director/stream_session.go index c114f06dc..489795a30 100644 --- a/pkg/director/stream_session.go +++ b/pkg/director/stream_session.go @@ -43,7 +43,6 @@ type StreamSession struct { cli *config.CLI bus *bus.Bus op *oatproxy.OATProxy - hls *media.M3U8 lp *livepeer.LivepeerSession repoDID string segmentChan chan struct{} @@ -108,7 +107,6 @@ func (ss *StreamSession) Start(ctx context.Context, notif *media.NewSegmentNotif } allRenditions = append([]renditions.Rendition{sourceRendition}, allRenditions...) allRenditions = append(allRenditions, renditions.AudioRendition) - ss.hls = media.NewM3U8(allRenditions) // ss.maybeStartS3Upload(ctx, notif.Segment.RepoDID) @@ -212,10 +210,6 @@ func (ss *StreamSession) NewSegment(ctx context.Context, notif *media.NewSegment }) } - ss.Go(ctx, func() error { - return ss.AddAudioOnlyHLSSegment(ctx, spseg, notif.Data) - }) - if ss.cli.Thumbnail { ss.Go(ctx, func() error { return ss.Thumbnail(ctx, spseg.Creator, notif) @@ -803,11 +797,6 @@ func (ss *StreamSession) Transcode(ctx context.Context, spseg *streamplace.Segme } func (ss *StreamSession) AddPlaybackSegment(ctx context.Context, spseg *streamplace.Segment, rendition string, seg *bus.Seg) error { - if seg.Published { - ss.Go(ctx, func() error { - return ss.AddToHLS(ctx, spseg, rendition, seg.Data) - }) - } ss.Go(ctx, func() error { return ss.AddToWebRTC(ctx, spseg, rendition, seg) }) @@ -824,67 +813,6 @@ func (ss *StreamSession) AddToWebRTC(ctx context.Context, spseg *streamplace.Seg return nil } -func (ss *StreamSession) AddToHLS(ctx context.Context, spseg *streamplace.Segment, rendition string, data []byte) error { - buf := bytes.Buffer{} - dur, err := media.MP4ToMPEGTS(ctx, bytes.NewReader(data), &buf) - if err != nil { - return fmt.Errorf("failed to convert MP4 to MPEG-TS: %w", err) - } - // newSeg := &streamplace.Segment{ - // LexiconTypeID: "place.stream.segment", - // Id: spseg.Id, - // Creator: spseg.Creator, - // StartTime: spseg.StartTime, - // Duration: &dur, - // Audio: spseg.Audio, - // Video: spseg.Video, - // SigningKey: spseg.SigningKey, - // } - aqt, err := aqtime.FromString(spseg.StartTime) - if err != nil { - return fmt.Errorf("failed to parse segment start time: %w", err) - } - log.Debug(ctx, "transmuxed to mpegts, adding to hls", "rendition", rendition, "size", buf.Len()) - rend, err := ss.hls.GetRendition(rendition) - if err != nil { - return fmt.Errorf("failed to get rendition: %w", err) - } - if err := rend.NewSegment(&media.Segment{ - Buf: &buf, - Duration: time.Duration(dur), - Time: aqt.Time(), - }); err != nil { - return fmt.Errorf("failed to create new segment: %w", err) - } - - return nil -} - -func (ss *StreamSession) AddAudioOnlyHLSSegment(ctx context.Context, spseg *streamplace.Segment, data []byte) error { - buf := bytes.Buffer{} - dur, err := media.MP4ToMPEGTSAudioOnly(ctx, bytes.NewReader(data), &buf) - if err != nil { - return fmt.Errorf("failed to convert MP4 to audio-only MPEG-TS: %w", err) - } - aqt, err := aqtime.FromString(spseg.StartTime) - if err != nil { - return fmt.Errorf("failed to parse segment start time: %w", err) - } - log.Debug(ctx, "transmuxed to audio-only mpegts, adding to hls", "size", buf.Len()) - rend, err := ss.hls.GetRendition(renditions.AudioRendition.Name) - if err != nil { - return fmt.Errorf("failed to get audio rendition: %w", err) - } - if err := rend.NewSegment(&media.Segment{ - Buf: &buf, - Duration: time.Duration(dur), - Time: aqt.Time(), - }); err != nil { - return fmt.Errorf("failed to create new audio segment: %w", err) - } - return nil -} - type XRPCClient interface { Do(ctx context.Context, method string, contentType string, path string, queryParams map[string]any, body any, out any) error } diff --git a/pkg/media/m3u8.go b/pkg/media/m3u8.go deleted file mode 100644 index 476367ff8..000000000 --- a/pkg/media/m3u8.go +++ /dev/null @@ -1,195 +0,0 @@ -package media - -import ( - "bytes" - "context" - "fmt" - "math" - "strings" - "sync" - "time" - - "github.com/google/uuid" - "stream.place/streamplace/pkg/log" - "stream.place/streamplace/pkg/renditions" -) - -// how many segments are served in the live playlist? -const LivePlaylistSize = 8 - -// how long should we keep old segments around? -const RetainSegmentSize = LivePlaylistSize * 3 - -const IndexM3U8 = "index.m3u8" - -type Segment struct { - MSN uint64 // media sequence number - Duration time.Duration - Buf *bytes.Buffer - Time time.Time - Closed bool - StartTS *uint64 - EndTS *uint64 -} - -type M3U8 struct { - curSeg uint64 - pendingSegments []*Segment - waits []chan struct{} - renditions []*M3U8Rendition -} - -type M3U8Rendition struct { - Rendition renditions.Rendition - Segments []*Segment - SegmentLock sync.RWMutex - MSN uint64 -} - -func NewM3U8(renditions renditions.Renditions) *M3U8 { - rends := []*M3U8Rendition{} - for _, r := range renditions { - mr := &M3U8Rendition{ - Rendition: r, - } - rends = append(rends, mr) - } - return &M3U8{ - curSeg: 0, - renditions: rends, - } -} - -func (r *M3U8Rendition) GetMediaLine(session string) string { - lines := []string{} - lines = append(lines, "#EXTM3U") - if r.Rendition.AudioOnly { - lines = append(lines, fmt.Sprintf("#EXT-X-STREAM-INF:BANDWIDTH=%d", r.Rendition.Bitrate)) - } else { - lines = append(lines, fmt.Sprintf("#EXT-X-STREAM-INF:BANDWIDTH=%d,RESOLUTION=%dx%d", r.Rendition.Bitrate, r.Rendition.Width, r.Rendition.Height)) - } - lines = append(lines, fmt.Sprintf("%s/%s?session=%s", r.Rendition.Name, IndexM3U8, session)) - return strings.Join(lines, "\n") -} - -func (r *M3U8Rendition) GetPlaylist(session string) []byte { - if session == "" { - uu, err := uuid.NewV7() - if err != nil { - panic(err) - } - session = uu.String() - } - r.SegmentLock.RLock() - defer r.SegmentLock.RUnlock() - // m.waitForStart() - lines := []string{} - lines = append(lines, "#EXTM3U") - lines = append(lines, "#EXT-X-VERSION:3") - startWith := len(r.Segments) - LivePlaylistSize - if startWith < 0 { - startWith = 0 - } - if len(r.Segments) == 0 { - return []byte{} - } - firstSeg := r.Segments[startWith] - lastSeg := r.Segments[len(r.Segments)-1] - targetDuration := int64(math.Round(lastSeg.Duration.Seconds())) - lines = append(lines, fmt.Sprintf("#EXT-X-MEDIA-SEQUENCE:%d", firstSeg.MSN)) - lines = append(lines, fmt.Sprintf("#EXT-X-DISCONTINUITY-SEQUENCE:%d", firstSeg.MSN)) - lines = append(lines, fmt.Sprintf("#EXT-X-TARGETDURATION:%d", targetDuration+1)) - lines = append(lines, "#EXT-X-INDEPENDENT-SEGMENTS") - lines = append(lines, "") - lastSegments := r.Segments[startWith:] - for _, seg := range lastSegments { - dur := seg.Duration - lines = append(lines, "#EXT-X-DISCONTINUITY") - lines = append(lines, fmt.Sprintf("#EXT-X-PROGRAM-DATE-TIME:%s", seg.Time.Format(time.RFC3339Nano))) - lines = append(lines, fmt.Sprintf("#EXTINF:%f,", dur.Seconds())) - lines = append(lines, fmt.Sprintf("segment%05d.ts?session=%s", seg.MSN, session)) - } - lines = append(lines, "") - return []byte(strings.Join(lines, "\n")) -} - -func (r *M3U8Rendition) GetSegment(session string, filename string) []byte { - r.SegmentLock.RLock() - defer r.SegmentLock.RUnlock() - for _, seg := range r.Segments { - if fmt.Sprintf("segment%05d.ts", seg.MSN) == filename { - return seg.Buf.Bytes() - } - } - return nil -} - -func (m *M3U8) GetMultivariantPlaylist(rendition string) []byte { - uu, err := uuid.NewV7() - if err != nil { - panic(err) - } - // m.waitForStart() - lines := []string{} - lines = append(lines, "#EXTM3U") - for _, r := range m.renditions { - if rendition == "" || r.Rendition.Name == rendition { - lines = append(lines, r.GetMediaLine(uu.String())) - } - } - return []byte(strings.Join(lines, "\n")) -} - -// needs to handle: -// - index.m3u8 -// - 720p/stream.m3u8 -// - 720p/segment00015.ts -func (m *M3U8) GetFile(str string, session string, rendition string) ([]byte, error) { - str = strings.TrimPrefix(str, "/") - if str == IndexM3U8 { - return m.GetMultivariantPlaylist(rendition), nil - } - parts := strings.Split(str, "/") - if len(parts) != 2 { - return nil, fmt.Errorf("invalid path") - } - rStr := parts[0] - fStr := parts[1] - rend, err := m.GetRendition(rStr) - if err != nil { - return nil, err - } - log.Debug(context.Background(), "m3u8 get file", "str", str, "session", session, "rend", rStr, "file", fStr) - if fStr == IndexM3U8 { - return rend.GetPlaylist(session), nil - } - seg := rend.GetSegment(session, fStr) - if seg == nil { - return nil, fmt.Errorf("segment not found") - } - return seg, nil -} - -func (r *M3U8Rendition) NewSegment(seg *Segment) error { - r.SegmentLock.Lock() - defer r.SegmentLock.Unlock() - seg.MSN = r.MSN - r.MSN += 1 - r.Segments = append(r.Segments, seg) - if len(r.Segments) > RetainSegmentSize { - // Calculate how many segments to remove - removeCount := len(r.Segments) - RetainSegmentSize - // Remove the oldest segments (from the front of the slice) - r.Segments = r.Segments[removeCount:] - } - return nil -} - -func (m *M3U8) GetRendition(rendition string) (*M3U8Rendition, error) { - for _, r := range m.renditions { - if r.Rendition.Name == rendition { - return r, nil - } - } - return nil, fmt.Errorf("rendition not found") -} diff --git a/pkg/media/media.go b/pkg/media/media.go index 196375314..9d51a3707 100644 --- a/pkg/media/media.go +++ b/pkg/media/media.go @@ -42,8 +42,6 @@ const StreamplaceMetadata = "cawg.metadata" type MediaManager struct { cli *config.CLI - hlsRunning map[string]*M3U8 - hlsRunningMut sync.Mutex liveWindows map[string]*livehls.Writer liveWindowsMut sync.Mutex httpPipes map[string]io.Writer @@ -125,7 +123,6 @@ func MakeMediaManager(ctx context.Context, cli *config.CLI, signer crypto.Signer } return &MediaManager{ cli: cli, - hlsRunning: map[string]*M3U8{}, liveWindows: map[string]*livehls.Writer{}, httpPipes: map[string]io.Writer{}, model: mod, diff --git a/pkg/media/segment_conv.go b/pkg/media/segment_conv.go index ebbb58073..9e590f7e0 100644 --- a/pkg/media/segment_conv.go +++ b/pkg/media/segment_conv.go @@ -5,372 +5,13 @@ import ( "fmt" "io" "strings" - "time" "github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" "golang.org/x/sync/errgroup" - "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/log" ) -// MP4ToMPEGTS converts an MP4 file with H264 video and Opus audio to an MPEG-TS file -// It reads from the provided reader and writes the converted MPEG-TS to the writer. -// The conversion is optimized for speed. -func MP4ToMPEGTS(ctx context.Context, input io.Reader, output io.Writer) (int64, error) { - ctx = log.WithLogValues(ctx, "func", "MP4ToMPEGTS") - pipelineStr := strings.Join([]string{ - "appsrc name=appsrc ! qtdemux name=demux", - "mpegtsmux name=mux ! appsink name=appsink sync=false", - fmt.Sprintf("demux.video_0 ! h264parse ! video/x-h264,stream-format=byte-stream ! %s name=videoqueue", constants.Queue2Big), - fmt.Sprintf("demux.audio_0 ! opusdec name=audioparse ! audioresample ! audiorate ! fdkaacenc name=audioenc ! %s name=audioqueue", constants.Queue2Big), - }, " ") - - pipeline, err := gst.NewPipelineFromString(pipelineStr) - if err != nil { - return 0, err - } - - mux, err := pipeline.GetElementByName("mux") - if err != nil { - return 0, err - } - muxVideoSinkPad := mux.GetRequestPad("sink_%d") - if muxVideoSinkPad == nil { - return 0, fmt.Errorf("failed to get video sink pad") - } - muxAudioSinkPad := mux.GetRequestPad("sink_%d") - if muxAudioSinkPad == nil { - return 0, fmt.Errorf("failed to get audio sink pad") - } - videoQueue, err := pipeline.GetElementByName("videoqueue") - if err != nil { - return 0, err - } - audioQueue, err := pipeline.GetElementByName("audioqueue") - if err != nil { - return 0, err - } - videoQueueSrcPad := videoQueue.GetStaticPad("src") - if videoQueueSrcPad == nil { - return 0, fmt.Errorf("failed to get video queue source pad") - } - audioQueueSrcPad := audioQueue.GetStaticPad("src") - if audioQueueSrcPad == nil { - return 0, fmt.Errorf("failed to get audio queue source pad") - } - - ok := videoQueueSrcPad.Link(muxVideoSinkPad) - if ok != gst.PadLinkOK { - return 0, fmt.Errorf("failed to link video queue source pad to mux video sink pad: %v", ok) - } - ok = audioQueueSrcPad.Link(muxAudioSinkPad) - if ok != gst.PadLinkOK { - return 0, fmt.Errorf("failed to link audio queue source pad to mux audio sink pad: %v", ok) - } - - // Get elements - appsrc, err := pipeline.GetElementByName("appsrc") - if err != nil { - return 0, err - } - appsink, err := pipeline.GetElementByName("appsink") - if err != nil { - return 0, err - } - - source := app.SrcFromElement(appsrc) - sink := app.SinkFromElement(appsink) - - // Set up source callbacks - source.SetCallbacks(&app.SourceCallbacks{ - NeedDataFunc: ReaderNeedDataIncremental(ctx, input), - EnoughDataFunc: func(self *app.Source) { - // Nothing to do here - }, - SeekDataFunc: func(self *app.Source, offset uint64) bool { - return false // We don't support seeking - }, - }) - - // Set up sink callbacks - sink.SetCallbacks(&app.SinkCallbacks{ - NewSampleFunc: WriterNewSample(ctx, output), - NewPrerollFunc: func(self *app.Sink) gst.FlowReturn { - return gst.FlowOK - }, - }) - - ctx, cancel := context.WithCancel(ctx) - defer func() { - cancel() - // Clean up - err = pipeline.SetState(gst.StateNull) - if err != nil { - log.Error(ctx, "failed to set pipeline state to null", "error", err) - } - }() - - go func() { - select { - case <-ctx.Done(): - return - case <-time.After(time.Second * 10): - log.Debug(ctx, "pipeline is taking too long to start, cancelling") - err := fmt.Errorf("pipeline is taking too long to start, cancelling") - pipeline.Error(err.Error(), err) - } - }() - - // Handle bus messages in a separate goroutine - errCh := make(chan error) - go func() { - err := HandleBusMessages(ctx, pipeline) - cancel() - errCh <- err - close(errCh) - }() - - // Start the pipeline - err = pipeline.SetState(gst.StatePlaying) - if err != nil { - return 0, fmt.Errorf("failed to set pipeline state to playing: %w", err) - } - - var durOk bool - var dur int64 - busErr := <-errCh - - if busErr == nil { - durOk, dur = pipeline.QueryDuration(gst.FormatTime) - if !durOk { - return 0, fmt.Errorf("failed to query duration") - } - } - - return dur, busErr -} - -// MP4ToMPEGTSAudioOnly converts an MP4 file to an audio-only MPEG-TS file, discarding video. -func MP4ToMPEGTSAudioOnly(ctx context.Context, input io.Reader, output io.Writer) (int64, error) { - ctx = log.WithLogValues(ctx, "func", "MP4ToMPEGTSAudioOnly") - pipelineStr := strings.Join([]string{ - "appsrc name=appsrc ! qtdemux name=demux", - "mpegtsmux name=mux ! appsink name=appsink sync=false", - "demux.audio_0 ! opusdec name=audioparse ! audioresample ! audiorate ! fdkaacenc name=audioenc ! queue name=audioqueue", - }, " ") - - pipeline, err := gst.NewPipelineFromString(pipelineStr) - if err != nil { - return 0, err - } - - mux, err := pipeline.GetElementByName("mux") - if err != nil { - return 0, err - } - muxAudioSinkPad := mux.GetRequestPad("sink_%d") - if muxAudioSinkPad == nil { - return 0, fmt.Errorf("failed to get audio sink pad") - } - audioQueue, err := pipeline.GetElementByName("audioqueue") - if err != nil { - return 0, err - } - audioQueueSrcPad := audioQueue.GetStaticPad("src") - if audioQueueSrcPad == nil { - return 0, fmt.Errorf("failed to get audio queue source pad") - } - ok := audioQueueSrcPad.Link(muxAudioSinkPad) - if ok != gst.PadLinkOK { - return 0, fmt.Errorf("failed to link audio queue source pad to mux audio sink pad: %v", ok) - } - - appsrc, err := pipeline.GetElementByName("appsrc") - if err != nil { - return 0, err - } - appsink, err := pipeline.GetElementByName("appsink") - if err != nil { - return 0, err - } - - source := app.SrcFromElement(appsrc) - sink := app.SinkFromElement(appsink) - - source.SetCallbacks(&app.SourceCallbacks{ - NeedDataFunc: ReaderNeedDataIncremental(ctx, input), - EnoughDataFunc: func(self *app.Source) { - }, - SeekDataFunc: func(self *app.Source, offset uint64) bool { - return false - }, - }) - - sink.SetCallbacks(&app.SinkCallbacks{ - NewSampleFunc: WriterNewSample(ctx, output), - NewPrerollFunc: func(self *app.Sink) gst.FlowReturn { - return gst.FlowOK - }, - }) - - ctx, cancel := context.WithCancel(ctx) - defer func() { - cancel() - err = pipeline.SetState(gst.StateNull) - if err != nil { - log.Error(ctx, "failed to set pipeline state to null", "error", err) - } - }() - - go func() { - select { - case <-ctx.Done(): - return - case <-time.After(time.Second * 10): - log.Debug(ctx, "pipeline is taking too long to start, cancelling") - err := fmt.Errorf("pipeline is taking too long to start, cancelling") - pipeline.Error(err.Error(), err) - } - }() - - errCh := make(chan error) - go func() { - err := HandleBusMessages(ctx, pipeline) - cancel() - errCh <- err - close(errCh) - }() - - err = pipeline.SetState(gst.StatePlaying) - if err != nil { - return 0, fmt.Errorf("failed to set pipeline state to playing: %w", err) - } - - var durOk bool - var dur int64 - busErr := <-errCh - - if busErr == nil { - durOk, dur = pipeline.QueryDuration(gst.FormatTime) - if !durOk { - return 0, fmt.Errorf("failed to query duration") - } - } - - return dur, busErr -} - -// MPEGTSToMP4 converts an MPEG-TS file with H264 video and Opus audio to an MP4 file. -// It reads from the provided reader and writes the converted MP4 to the writer. -func MPEGTSToMP4(ctx context.Context, input io.Reader, output io.Writer) error { - ctx = log.WithLogValues(ctx, "func", "MPEGTSToMP4") - pipelineStr := strings.Join([]string{ - "appsrc name=appsrc ! tsdemux name=demux", - "mp4mux name=mux ! appsink sync=false name=appsink", - "demux.video_0_0100 ! h264parse ! video/x-h264,stream-format=avc ! queue name=videoqueue", - "demux.audio_0_0101 ! opusdec ! opusenc ! queue name=audioqueue", - }, " ") - - pipeline, err := gst.NewPipelineFromString(pipelineStr) - if err != nil { - return err - } - - mux, err := pipeline.GetElementByName("mux") - if err != nil { - return err - } - muxVideoSinkPad := mux.GetRequestPad("video_%u") - if muxVideoSinkPad == nil { - return fmt.Errorf("failed to get video sink pad") - } - muxAudioSinkPad := mux.GetRequestPad("audio_%u") - if muxAudioSinkPad == nil { - return fmt.Errorf("failed to get audio sink pad") - } - videoQueue, err := pipeline.GetElementByName("videoqueue") - if err != nil { - return err - } - audioQueue, err := pipeline.GetElementByName("audioqueue") - if err != nil { - return err - } - videoQueueSrcPad := videoQueue.GetStaticPad("src") - if videoQueueSrcPad == nil { - return fmt.Errorf("failed to get video queue source pad") - } - audioQueueSrcPad := audioQueue.GetStaticPad("src") - if audioQueueSrcPad == nil { - return fmt.Errorf("failed to get audio queue source pad") - } - - ok := videoQueueSrcPad.Link(muxVideoSinkPad) - if ok != gst.PadLinkOK { - return fmt.Errorf("failed to link video queue source pad to mux video sink pad: %v", ok) - } - ok = audioQueueSrcPad.Link(muxAudioSinkPad) - if ok != gst.PadLinkOK { - return fmt.Errorf("failed to link audio queue source pad to mux audio sink pad: %v", ok) - } - - // Get elements - appsrc, err := pipeline.GetElementByName("appsrc") - if err != nil { - return err - } - appsink, err := pipeline.GetElementByName("appsink") - if err != nil { - return err - } - - source := app.SrcFromElement(appsrc) - sink := app.SinkFromElement(appsink) - - // Set up source callbacks - source.SetCallbacks(&app.SourceCallbacks{ - NeedDataFunc: ReaderNeedDataIncremental(ctx, input), - EnoughDataFunc: func(self *app.Source) { - // Nothing to do here - }, - SeekDataFunc: func(self *app.Source, offset uint64) bool { - return false // We don't support seeking - }, - }) - - // Set up sink callbacks - sink.SetCallbacks(&app.SinkCallbacks{ - NewSampleFunc: WriterNewSample(ctx, output), - NewPrerollFunc: func(self *app.Sink) gst.FlowReturn { - return gst.FlowOK - }, - }) - - ctx, cancel := context.WithCancel(ctx) - defer cancel() - - busErr := make(chan error) - go func() { - err := HandleBusMessages(ctx, pipeline) - busErr <- err - }() - - err = pipeline.SetState(gst.StatePlaying) - if err != nil { - return fmt.Errorf("failed to set pipeline state to playing: %w", err) - } - - defer func() { - err = pipeline.SetState(gst.StateNull) - if err != nil { - log.Error(ctx, "failed to set pipeline state to null", "error", err) - } - }() - - return <-busErr -} - // Splits out video into MPEG-TS and audio into MP4 (to be recombined after transcoding) func MP4ToMPEGTSVideoMP4Audio(ctx context.Context, input io.Reader, videoOutput io.Writer, audioOutput io.Writer) error { ctx = log.WithLogValues(ctx, "func", "MP4ToMPEGTSVideoMP4Audio") diff --git a/pkg/media/segment_conv_test.go b/pkg/media/segment_conv_test.go index 90f6ee8ee..be42a50ac 100644 --- a/pkg/media/segment_conv_test.go +++ b/pkg/media/segment_conv_test.go @@ -3,7 +3,6 @@ package media import ( "bytes" "context" - "io" "math/rand" "os" "testing" @@ -11,123 +10,8 @@ import ( "github.com/stretchr/testify/require" "golang.org/x/sync/errgroup" - "stream.place/streamplace/test/remote" ) -func TestMP4ToMPEGTS(t *testing.T) { - withNoGSTLeaks(t, func() { - for _, file := range []string{ - getFixture("sample-segment.mp4"), - getFixture("short-video.mp4"), - remote.RemoteFixture("82d20ee62b02f1c3a727b3001f1fa939afb757f9f205fa438d7b5753e1253eef/2026-04-11T22-39-41-861Z-packetize-input-019d7eb3-6f24-776c-ba1b-2f909a2379d7.mp4")} { - // Open input file - inputFile, err := os.Open(file) - require.NoError(t, err) - defer inputFile.Close() - - // Create a buffer for output - buf := bytes.Buffer{} - - bs, err := io.ReadAll(inputFile) - require.NoError(t, err) - // Create temporary output file - - g, _ := errgroup.WithContext(context.Background()) - for range streamplaceTestCount { - g.Go(func() error { - _, err := MP4ToMPEGTS(context.Background(), bytes.NewReader(bs), &buf) - return err - }) - } - err = g.Wait() - require.NoError(t, err) - // Convert MPEG-TS to MP4 - - // Convert MP4 to MPEG-TS - dur, err := MP4ToMPEGTS(context.Background(), bytes.NewReader(bs), &buf) - require.NoError(t, err) - require.Greater(t, dur, int64(0), "Duration should be greater than 0") - - // Verify buffer has content - require.Greater(t, buf.Len(), 0, "Output buffer should not be empty") - } - }) -} - -func TestMP4ToMPEGTSInvalid(t *testing.T) { - withNoGSTLeaks(t, func() { - - inputFile, err := os.Open(getFixture("sample-segment.mpegts")) - require.NoError(t, err) - defer inputFile.Close() - - buf := bytes.Buffer{} - - bs, err := io.ReadAll(inputFile) - require.NoError(t, err) - - // Convert MP4 to MPEG-TS - _, err = MP4ToMPEGTS(context.Background(), bytes.NewReader(bs), &buf) - require.ErrorContains(t, err, "This file is invalid and cannot be played") - }) -} - -// func TestNoRealtime(t *testing.T) { -// gst.Init(nil) - -// // Open input file -// inputFile, err := os.Open(getFixture("5sec.mp4")) -// require.NoError(t, err) -// defer inputFile.Close() - -// // Create a buffer for output -// tsBuf := bytes.Buffer{} - -// // Convert MP4 to MPEG-TS -// start := time.Now() -// dur, err := MP4ToMPEGTS(context.Background(), inputFile, &tsBuf) -// require.NoError(t, err) -// require.Greater(t, dur, int64(0), "Duration should be greater than 0") -// elapsed := time.Since(start) -// require.Less(t, elapsed, 4*time.Second, "MP4 to MPEG-TS conversion should take less than 4 seconds") -// require.Greater(t, tsBuf.Len(), 0, "MPEG-TS buffer should not be empty") - -// // Convert back to MP4 -// mp4Buf := bytes.Buffer{} -// start = time.Now() -// err = MPEGTSToMP4(context.Background(), bytes.NewReader(tsBuf.Bytes()), &mp4Buf) -// require.NoError(t, err) -// elapsed = time.Since(start) -// require.Less(t, elapsed, 4*time.Second, "MPEG-TS to MP4 conversion should take less than 4 seconds") -// require.Greater(t, mp4Buf.Len(), 0, "MP4 buffer should not be empty") -// } - -func TestMPEGTSToMP4(t *testing.T) { - withNoGSTLeaks(t, func() { - - // Open input file - inputFile, err := os.Open(getFixture("sample-segment.mpegts")) - require.NoError(t, err) - defer inputFile.Close() - bs, err := io.ReadAll(inputFile) - require.NoError(t, err) - // Create temporary output file - buf := bytes.Buffer{} - - g, _ := errgroup.WithContext(context.Background()) - for i := 0; i < streamplaceTestCount; i++ { - g.Go(func() error { - return MPEGTSToMP4(context.Background(), bytes.NewReader(bs), &buf) - }) - } - err = g.Wait() - // Convert MPEG-TS to MP4 - - require.NoError(t, err) - require.Greater(t, buf.Len(), 0, "Output file should not be empty") - }) -} - func TestMP4ToMPEGTSVideoMP4Audio(t *testing.T) { withNoGSTLeaks(t, func() { diff --git a/pkg/media/segmenter_hls.go b/pkg/media/segmenter_hls.go deleted file mode 100644 index 68649bf17..000000000 --- a/pkg/media/segmenter_hls.go +++ /dev/null @@ -1,312 +0,0 @@ -package media - -import ( - "bytes" - "context" - "fmt" - "strings" - "sync" - "time" - - "github.com/go-gst/go-gst/gst" - "github.com/go-gst/go-gst/gst/app" - "stream.place/streamplace/pkg/log" -) - -func (mm *MediaManager) ToHLS(ctx context.Context, user string, rendition string, m3u8 *M3U8) error { - ctx = log.WithLogValues(ctx, "GStreamerFunc", "ToHLS", "rendition", rendition) - - pipelineSlice := []string{ - "h264parse name=videoparse", - "opusdec use-inband-fec=true name=audioparse ! audioresample ! audiorate ! fdkaacenc name=audioenc", - } - - pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n")) - if err != nil { - return fmt.Errorf("error creating ToHLS pipeline: %w", err) - } - - outputQueue, done, err := ConcatStream(ctx, pipeline, user, rendition, mm.bus) - if err != nil { - return fmt.Errorf("failed to get output queue: %w", err) - } - - videoParse, err := pipeline.GetElementByName("videoparse") - if err != nil { - return fmt.Errorf("failed to get video sink element from pipeline: %w", err) - } - err = outputQueue.Link(videoParse) - if err != nil { - return fmt.Errorf("failed to link output queue to video parse: %w", err) - } - - audioParse, err := pipeline.GetElementByName("audioparse") - if err != nil { - return fmt.Errorf("failed to get audio parse element from pipeline: %w", err) - } - err = outputQueue.Link(audioParse) - if err != nil { - return fmt.Errorf("failed to link output queue to audio parse: %w", err) - } - - splitmuxsink, err := gst.NewElementWithProperties("splitmuxsink", map[string]any{ - "name": "mux", - "async-finalize": true, - "sink-factory": "appsink", - "muxer-factory": "mpegtsmux", - "max-size-bytes": 1, - }) - if err != nil { - return err - } - - r, err := m3u8.GetRendition(rendition) - if err != nil { - return fmt.Errorf("failed to get rendition: %w", err) - } - defer func() { r = nil }() - ps := NewPendingSegments(r) - defer func() { ps = nil }() - - p := splitmuxsink.GetRequestPad("video") - if p == nil { - return fmt.Errorf("failed to get video pad") - } - p = splitmuxsink.GetRequestPad("audio_%u") - if p == nil { - return fmt.Errorf("failed to get audio pad") - } - - err = pipeline.Add(splitmuxsink) - if err != nil { - return fmt.Errorf("error adding splitmuxsink to ToHLS pipeline: %w", err) - } - - videoparse, err := pipeline.GetElementByName("videoparse") - if err != nil { - return fmt.Errorf("error getting videoparse from ToHLS pipeline: %w", err) - } - err = videoparse.Link(splitmuxsink) - if err != nil { - return fmt.Errorf("error linking videoparse to splitmuxsink: %w", err) - } - - audioenc, err := pipeline.GetElementByName("audioenc") - if err != nil { - return fmt.Errorf("error getting audioenc from ToHLS pipeline: %w", err) - } - err = audioenc.Link(splitmuxsink) - if err != nil { - return fmt.Errorf("error linking audioenc to splitmuxsink: %w", err) - } - - ctx, cancel := context.WithCancel(ctx) - - go func() { - select { - case <-ctx.Done(): - return - case <-done: - cancel() - } - }() - - _, err = splitmuxsink.Connect("sink-added", func(split, sinkEle *gst.Element) { - log.Debug(ctx, "hls-check sink-added") - vf, err := ps.GetNextSegment(ctx) - if err != nil { - panic(err) - } - appsink := app.SinkFromElement(sinkEle) - appsink.SetCallbacks(&app.SinkCallbacks{ - NewSampleFunc: WriterNewSample(ctx, vf.Buf), - EOSFunc: func(sink *app.Sink) { - log.Debug(ctx, "hls-check Segment EOS", "buf", vf.Buf.Len()) - ps.CloseSegment(ctx, vf) - }, - }) - }) - if err != nil { - return fmt.Errorf("failed to add hls-check to sink: %w", err) - } - - onPadAdded := func(element *gst.Element, pad *gst.Pad) { - caps := pad.GetCurrentCaps() - if caps == nil { - fmt.Println("Unable to get pad caps") - return - } - - log.Debug(ctx, "New pad added", "pad", pad.GetName(), "caps", caps.String()) - - structure := caps.GetStructureAt(0) - if structure == nil { - fmt.Println("Unable to get structure from caps") - return - } - - name := structure.Name() - fmt.Printf("Structure Name: %s\n", name) - } - - _, err = splitmuxsink.Connect("pad-added", onPadAdded) - if err != nil { - return fmt.Errorf("failed to add pad: %w", err) - } - - defer cancel() - go func() { - err := HandleBusMessagesCustom(ctx, pipeline, func(msg *gst.Message) { - switch msg.Type() { - case gst.MessageElement: - structure := msg.GetStructure() - name := structure.Name() - if name == "splitmuxsink-fragment-opened" { - runningTime, err := structure.GetValue("running-time") - if err != nil { - log.Debug(ctx, "splitmuxsink-fragment-opened error", "error", err) - cancel() - } - runningTimeInt, ok := runningTime.(uint64) - if !ok { - log.Warn(ctx, "splitmuxsink-fragment-opened not a uint64") - cancel() - } - log.Debug(ctx, "hls-check splitmuxsink-fragment-opened", "runningTime", runningTimeInt) - if err := ps.FragmentOpened(ctx, runningTimeInt); err != nil { - log.Debug(ctx, "fragment open error", "error", err) - cancel() - } - } - if name == "splitmuxsink-fragment-closed" { - runningTime, err := structure.GetValue("running-time") - if err != nil { - log.Debug(ctx, "splitmuxsink-fragment-closed error", "error", err) - cancel() - } - runningTimeInt, ok := runningTime.(uint64) - if !ok { - log.Warn(ctx, "splitmuxsink-fragment-closed not a uint64") - cancel() - } - log.Debug(ctx, "hls-check splitmuxsink-fragment-closed", "runningTime", runningTimeInt) - if err := ps.FragmentClosed(ctx, runningTimeInt); err != nil { - log.Debug(ctx, "fragment close error", "error", err) - cancel() - } - } - } - }) - if err != nil { - log.Log(ctx, "pipeline error", "error", err) - } - cancel() - }() - - // Start the pipeline - if err := pipeline.SetState(gst.StatePlaying); err != nil { - return fmt.Errorf("error setting pipeline state: %w", err) - } - - <-ctx.Done() - - if err := pipeline.BlockSetState(gst.StateNull); err != nil { - return fmt.Errorf("error setting pipeline state: %w", err) - } - - return nil -} - -type PendingSegments struct { - segments []*Segment - lock sync.Mutex - rendition *M3U8Rendition -} - -func NewPendingSegments(rendition *M3U8Rendition) *PendingSegments { - return &PendingSegments{ - segments: []*Segment{}, - lock: sync.Mutex{}, - rendition: rendition, - } -} - -func (ps *PendingSegments) GetNextSegment(ctx context.Context) (*Segment, error) { - ps.lock.Lock() - defer ps.lock.Unlock() - log.Debug(ctx, "next segment") - seg := &Segment{ - Buf: &bytes.Buffer{}, - Time: time.Now(), - Closed: false, - } - ps.segments = append(ps.segments, seg) - return seg, nil -} - -func (ps *PendingSegments) CloseSegment(ctx context.Context, seg *Segment) { - ps.lock.Lock() - defer ps.lock.Unlock() - log.Debug(ctx, "close segment", "MSN", seg.MSN) - seg.Closed = true - if err := ps.checkSegments(ctx); err != nil { - log.Debug(ctx, "faile to check segments segment") - } -} - -func (ps *PendingSegments) FragmentOpened(ctx context.Context, t uint64) error { - ps.lock.Lock() - defer ps.lock.Unlock() - log.Debug(ctx, "fragment opened", "time", t) - if len(ps.segments) == 0 { - return fmt.Errorf("no pending segments") - } - for _, seg := range ps.segments { - if seg.StartTS == nil { - seg.StartTS = &t - break - } - } - if err := ps.checkSegments(ctx); err != nil { - return fmt.Errorf("failed to check segments: %w", err) - } - return nil -} - -func (ps *PendingSegments) FragmentClosed(ctx context.Context, t uint64) error { - ps.lock.Lock() - defer ps.lock.Unlock() - log.Debug(ctx, "fragment closed", "time", t) - if len(ps.segments) == 0 { - return fmt.Errorf("no pending segments") - } - for _, seg := range ps.segments { - if seg.EndTS == nil { - seg.EndTS = &t - dur := *seg.EndTS - *seg.StartTS - seg.Duration = time.Duration(dur) - break - } - } - if err := ps.checkSegments(ctx); err != nil { - return fmt.Errorf("failed to check segments: %w", err) - } - return nil -} - -// the tricky piece of the design here is that we need to expect GetNextSegment, -// CloseSegment, FragmentOpened, and FragmentClosed to be called in any order. So -// all of those functions call this one, and it checks if we have the necessary information -// to finalize a segment and add it to our playlist. -// only call if you're holding ps.lock! -func (ps *PendingSegments) checkSegments(ctx context.Context) error { - pending := ps.segments[0] - if pending.StartTS != nil && pending.EndTS != nil && pending.Closed { - if err := ps.rendition.NewSegment(pending); err != nil { - return fmt.Errorf("failed to add new segment: %w", err) - } - log.Debug(ctx, "finalizing segment", "MSN", pending.MSN) - ps.segments = ps.segments[1:] - } - return nil -}