Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
9.0 kB · 276 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277package media
import ( "bytes" "context" "fmt" "strings" "sync/atomic"
"github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" "stream.place/streamplace/pkg/bus" "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/log")
// silly technique to avoid leaking padsfunc doNothing(self *gst.Element, pad *gst.Pad) {}
// doNothing for signals without a pad argumentfunc doNothingElement(self *gst.Element) {}
// Function for demuxing a single segment. Needs to be handled very carefully.// In particular: users of this MUST cancel the passed context when they're// done with the bin.func ConcatDemuxBin(ctx context.Context, seg *bus.Seg, doH264Parse bool) (*gst.Bin, error) { ctx = log.WithLogValues(ctx, "func", "ConcatDemuxBin") bin := gst.NewBin("seg-demux-bin")
appSrc, err := gst.NewElementWithProperties("appsrc", map[string]interface{}{ "name": "concat-appsrc", }) if err != nil { return nil, fmt.Errorf("failed to create appsrc element: %w", err) } err = bin.Add(appSrc) if err != nil { return nil, fmt.Errorf("failed to add appsrc to bin: %w", err) }
demux, err := gst.NewElementWithProperties("qtdemux", map[string]interface{}{ "name": "concat-demux", }) if err != nil { return nil, fmt.Errorf("failed to create qtdemux element: %w", err) } err = bin.Add(demux) if err != nil { return nil, fmt.Errorf("failed to add qtdemux to bin: %w", err) }
err = appSrc.Link(demux) if err != nil { return nil, fmt.Errorf("failed to link appsrc to qtdemux: %w", err) }
tmpl := demux.GetPadTemplates() if tmpl == nil { return nil, fmt.Errorf("pad templates not found") }
mq, err := gst.NewElementWithProperties("multiqueue", map[string]interface{}{ "name": "concat-demux-multiqueue", "max-size-time": constants.QueueMaxSizeTime, "max-size-bytes": constants.QueueMaxSizeBytes, "max-size-buffers": constants.QueueMaxSizeBuffers, }) if err != nil { return nil, fmt.Errorf("failed to create multiqueue element: %w", err) } err = bin.Add(mq) if err != nil { return nil, fmt.Errorf("failed to add multiqueue to bin: %w", err) } // err = mq.SetProperty("max-size-time", uint64(200000000000)) // if err != nil { // return nil, fmt.Errorf("failed to set max-size-time: %w", err) // } // err = mq.SetProperty("max-size-bytes", uint(1048576000)) // if err != nil { // return nil, fmt.Errorf("failed to set max-size-bytes: %w", err) // } // err = mq.SetProperty("max-size-buffers", uint(500)) // if err != nil { // return nil, fmt.Errorf("failed to set max-size-buffers: %w", err) // }
opusparse, err := gst.NewElementWithProperties("opusparse", map[string]interface{}{ "name": "concat-demux-opusparse", "disable-passthrough": true, }) if err != nil { return nil, fmt.Errorf("failed to create opusparse element: %w", err) } err = bin.Add(opusparse) if err != nil { return nil, fmt.Errorf("failed to add opusparse to bin: %w", err) } opusparseSinkPad := opusparse.GetStaticPad("sink") if opusparseSinkPad == nil { return nil, fmt.Errorf("failed to get opusparse sink pad") } opusparseSrcPad := opusparse.GetStaticPad("src") if opusparseSrcPad == nil { return nil, fmt.Errorf("failed to get opusparse source pad") }
mqVideoSink := mq.GetRequestPad("sink_%u") if mqVideoSink == nil { return nil, fmt.Errorf("video sink pad not found") }
mqAudioSink := mq.GetRequestPad("sink_%u") if mqAudioSink == nil { return nil, fmt.Errorf("audio sink pad not found") }
mqVideoSrc := mq.GetStaticPad("src_0") if mqVideoSrc == nil { return nil, fmt.Errorf("video source pad not found") }
mqAudioSrc := mq.GetStaticPad("src_1") if mqAudioSrc == nil { return nil, fmt.Errorf("audio source pad not found") }
linked := mqAudioSrc.Link(opusparseSinkPad) if linked != gst.PadLinkOK { return nil, fmt.Errorf("failed to link opusparse sink pad to mq audio sink pad") }
var videoGhost *gst.GhostPad if doH264Parse { h264parse, err := gst.NewElementWithProperties("h264parse", map[string]interface{}{ "name": "concat-demux-h264parse", "config-interval": -1, "disable-passthrough": true, }) if err != nil { return nil, fmt.Errorf("failed to create h264parse element: %w", err) } err = bin.Add(h264parse) if err != nil { return nil, fmt.Errorf("failed to add h264parse to bin: %w", err) } h264parseSinkPad := h264parse.GetStaticPad("sink") if h264parseSinkPad == nil { return nil, fmt.Errorf("failed to get h264parse sink pad") } h264parseSrcPad := h264parse.GetStaticPad("src") if h264parseSrcPad == nil { return nil, fmt.Errorf("failed to get h264parse source pad") } linked := mqVideoSrc.Link(h264parseSinkPad) if linked != gst.PadLinkOK { return nil, fmt.Errorf("failed to link h264parse sink pad to mq video sink pad") }
videoGhost = gst.NewGhostPad("video_0", h264parseSrcPad) if videoGhost == nil { return nil, fmt.Errorf("failed to create video ghost pad") } } else { videoGhost = gst.NewGhostPad("video_0", mqVideoSrc) if videoGhost == nil { return nil, fmt.Errorf("failed to create video ghost pad") } }
audioGhost := gst.NewGhostPad("audio_0", opusparseSrcPad) if audioGhost == nil { return nil, fmt.Errorf("failed to create audio ghost pad") }
needed := 2 videoLinked := false audioLinked := false
// The signal handlers below are swapped for no-ops once they've done // their job (or the ctx ends, whichever comes first) so their closures // release the mq pads — and, via the pads' sticky events, tag lists and // samples — that they capture; without the swap those leak. The swap // happens on the ctx goroutine while the demux's dispatch thread reads // the same slot, so both sides go through an atomic (a plain assignment // here is a data race; a mutex held across dispatch risks blocking the // swap behind a stuck handler, pinning the closures). var padAdded atomic.Value padAdded.Store(func(self *gst.Element, pad *gst.Pad) { var downstreamPad *gst.Pad if strings.HasPrefix(pad.GetName(), "video_") { downstreamPad = mqVideoSink videoLinked = true } else if strings.HasPrefix(pad.GetName(), "audio_") { downstreamPad = mqAudioSink audioLinked = true } else { log.Error(ctx, "unknown pad", "name", pad.GetName(), "direction", pad.GetDirection()) // cancel() return } ret := pad.Link(downstreamPad) if ret != gst.PadLinkOK { log.Error(ctx, "failed to link demux to downstream pad", "name", pad.GetName(), "direction", pad.GetDirection(), "error", ret) // cancel() return } needed-- if needed == 0 { padAdded.Store(doNothing) } }) outerPadAdded := func(self *gst.Element, pad *gst.Pad) { padAdded.Load().(func(self *gst.Element, pad *gst.Pad))(self, pad) }
// Both branches are wired up before the demux runs, but the segment may // not have both tracks: a stream whose encoder is shedding everything but // keyframes under bandwidth pressure can produce a segment with video and // no audio (or, in principle, the reverse). The demux only EOSes pads it // created, so a branch whose pad never appears would never see data OR // EOS — its sink never finishes, the pipeline never completes, and the // caller hangs until its timeout. When the demux declares it's done making // pads, flush any branch left orphaned with an EOS so the pipeline can // drain. Fires on the demux's streaming thread, same as pad-added, so the // linked flags need no synchronization. var noMorePads atomic.Value noMorePads.Store(func(self *gst.Element) { if !videoLinked { log.Warn(ctx, "segment has no video track; completing video branch with EOS") mqVideoSink.SendEvent(gst.NewEOSEvent()) } if !audioLinked { log.Warn(ctx, "segment has no audio track; completing audio branch with EOS") mqAudioSink.SendEvent(gst.NewEOSEvent()) } noMorePads.Store(doNothingElement) }) outerNoMorePads := func(self *gst.Element) { noMorePads.Load().(func(self *gst.Element))(self) }
// Necessary to avoid leaking `mqVideoSink` and `mqAudioSink` from the // handlers in the case where we hit invalid data and they never fire. go func() { <-ctx.Done() padAdded.Store(doNothing) noMorePads.Store(doNothingElement) }()
_, err = demux.Connect("pad-added", outerPadAdded) if err != nil { return nil, fmt.Errorf("failed to connect demux pad-added signal: %w", err) } _, err = demux.Connect("no-more-pads", outerNoMorePads) if err != nil { return nil, fmt.Errorf("failed to connect demux no-more-pads signal: %w", err) }
ok := bin.AddPad(videoGhost.Pad) if !ok { return nil, fmt.Errorf("failed to add video ghost pad to bin") }
ok = bin.AddPad(audioGhost.Pad) if !ok { return nil, fmt.Errorf("failed to add audio ghost pad to bin") }
src := app.SrcFromElement(appSrc) src.SetCallbacks(&app.SourceCallbacks{ NeedDataFunc: ReaderNeedDataIncremental(ctx, bytes.NewReader(seg.Data)), })
return bin, nil}