Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
24 kB · 706 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707package media
import ( "context" "errors" "fmt" "io" "strconv" "strings" "sync" "sync/atomic" "time"
"github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" "go.opentelemetry.io/otel/trace" "stream.place/streamplace/pkg/log")
var vodPipelineTracer = otel.Tracer("vod-pipeline")
// elementaryStreamCaps lists the caps names that come out of a// demuxer as already-elementary (one codec per pad). We use this set// in parsebin's autoplug-continue callback to tell parsebin "stop// here, don't plug another parser" — wireVideoChain and// wireAudioChain plug all the parsing/decoding we need downstream.// Keep this in sync with the codecs handled there.var elementaryStreamCaps = map[string]bool{ "video/x-h264": true, "video/x-h265": true, // unsupported; we want a clean ErrUnsupportedCodec, not a crash "audio/mpeg": true, // AAC (only — MP3 has no decoder linked, will fail in audioChainSpec) "audio/x-opus": true,}
// ErrUnsupportedCodec is returned by RunVODPipeline when the input// contains a stream we can't process. Currently: any non-h264 video,// or audio that isn't AAC or Opus. Opus is the only audio codec we// transcode (to AAC via fdkaacenc); AAC passes through.var ErrUnsupportedCodec = errors.New("unsupported codec")
// ErrPipelineStalled is returned when the pipeline produces no output for// vodStallTimeout. The motivating case: a demuxer that reaches PLAYING but// never produces a usable pad (e.g. the parsebin/aacparse recursion, or a// codec gstreamer can't handle), leaving RunVODPipeline blocked forever on// the bus. The watchdog converts that silent hang into this error.var ErrPipelineStalled = errors.New("vod pipeline stalled")
const ( // vodStallTimeout bounds how long the pipeline may produce zero new // output before we treat it as hung, cancel the context, and return // ErrPipelineStalled. A healthy stream emits its first fragment within // well under a second of PLAYING and fragments steadily thereafter, so // 30s of total silence means something wedged upstream. vodStallTimeout = 30 * time.Second // vodProgressInterval is the heartbeat/stall-check cadence. Each tick // logs a progress line and checks for a stall. vodProgressInterval = 5 * time.Second)
// progressWriter wraps the appsink output writer to track how much fMP4 has// been emitted and when the last write happened. The watchdog reads these// to log a heartbeat and to detect a stall. All fields are touched from the// gstreamer streaming thread (Write) and the watchdog goroutine (the// readers), so access is atomic.type progressWriter struct { w io.Writer bytes atomic.Int64 samples atomic.Int64 lastNano atomic.Int64 // UnixNano of the most recent non-empty write stalled atomic.Bool}
func newProgressWriter(w io.Writer) *progressWriter { p := &progressWriter{w: w} p.lastNano.Store(time.Now().UnixNano()) return p}
func (p *progressWriter) Write(b []byte) (int, error) { n, err := p.w.Write(b) if n > 0 { p.bytes.Add(int64(n)) p.samples.Add(1) p.lastNano.Store(time.Now().UnixNano()) } return n, err}
// resetTimer marks "now" as the last-progress point. Called when the// pipeline reaches PLAYING so the stall clock measures from the moment// output could legitimately start, not from pipeline construction.func (p *progressWriter) resetTimer() { p.lastNano.Store(time.Now().UnixNano()) }
func (p *progressWriter) idle() time.Duration { return time.Since(time.Unix(0, p.lastNano.Load()))}
// VODResult bundles probe-derived metadata that the caller needs for// downstream record creation. Populated incrementally by the pad-added// handler (video/audio codec + dimensions/rate from parsebin caps) and// finalized after EOS with a duration query against the pipeline.type VODResult struct { // DurationMS is the duration of the source as reported by gstreamer // after EOS, in milliseconds. Zero if the duration query failed // (rare; bug or live source). DurationMS int64 // Video is the video track's probe metadata. nil if no video pad // was wired (only happens when the input is audio-only, which the // pipeline rejects today via "no video stream found"). Video *VODVideoTrack // Audio is the audio track's probe metadata. nil if the input had // no audio. Audio *VODAudioTrack}
// VODVideoTrack is per-track probe metadata for a video stream.// Codec is the gstreamer caps name without the leading "video/" (e.g.// "x-h264"); width/height are pixels; framerate is the parsebin-// reported frame rate in num/den form (often non-integer for variable-// frame-rate content).type VODVideoTrack struct { Codec string Width int Height int FPSNum int FPSDen int}
// VODAudioTrack is per-track probe metadata for an audio stream. Codec// is the gstreamer caps name (e.g. "x-opus", "mpeg"). For audio/mpeg// callers should also consult MPEGVersion to disambiguate AAC vs MP3.type VODAudioTrack struct { Codec string Rate int Channels int MPEGVersion int}
// RunVODPipeline runs the gstreamer side of VOD processing: read the// source media from src (size bytes total), parse it via parsebin,// re-mux it to fragmented MP4, and write that fMP4 to out. Audio is// transcoded to AAC via fdkaacenc when not already AAC; video must be// h264 — anything else returns ErrUnsupportedCodec.//// The function blocks until the input is fully consumed or an error// occurs. out is written from a gstreamer streaming thread, so it must// be safe for concurrent use with this goroutine (typically an io.Pipe// or a buffered writer fed by a single reader).//// On success, the returned VODResult carries probe metadata gathered// during the run (track codec/dimensions, total duration) for use by// the caller's downstream record-creation logic.func RunVODPipeline(ctx context.Context, src io.ReaderAt, size int64, out io.Writer) (VODResult, error) { ctx, cancel := context.WithCancel(ctx) defer cancel() ctx = log.WithLogValues(ctx, "func", "RunVODPipeline")
ctx, span := vodPipelineTracer.Start(ctx, "vod.RunVODPipeline", trace.WithAttributes( attribute.Int64("source_size_bytes", size), )) defer span.End()
log.Debug(ctx, "creating pipeline", "source_size", size)
var result VODResult
// Tracks fMP4 bytes flowing out of the appsink so the watchdog below // can emit progress and notice a stall. prog := newProgressWriter(out)
pipeline, err := gst.NewPipeline("vod-process") if err != nil { span.RecordError(err) return result, fmt.Errorf("new pipeline: %w", err) } defer func() { if setErr := pipeline.BlockSetState(gst.StateNull); setErr != nil { log.Error(ctx, "failed to set pipeline to null", "error", setErr) } }()
srcBin, err := RandomAccessSrcBin(ctx, "vod-src", src, size) if err != nil { return result, fmt.Errorf("source bin: %w", err) } if err := pipeline.Add(srcBin.Element); err != nil { return result, fmt.Errorf("add src bin: %w", err) }
parsebin, err := gst.NewElementWithProperties("parsebin", map[string]any{ "name": "vod-parsebin", "expose-all-streams": true, }) if err != nil { return result, fmt.Errorf("create parsebin: %w", err) } if err := pipeline.Add(parsebin); err != nil { return result, fmt.Errorf("add parsebin: %w", err) } if err := srcBin.Link(parsebin); err != nil { return result, fmt.Errorf("link src bin -> parsebin: %w", err) }
// Tell parsebin's autoplug machinery to stop at the elementary-stream // pads coming out of the demuxer. We plug our own h264parse / aacparse // / mpegaudioparse / opusdec etc. in wireVideoChain and wireAudioChain // downstream, so parsebin doesn't need to add another parser on top. // // Returning FALSE here is also our workaround for a parsebin bug // where audio/mpeg (AAC) triggers an unbounded recursion: aacparse-0 // fixes its src caps on the first frame, the resulting "caps" notify // re-enters parsebin's pad_added_cb, which plugs aacparse-1, whose // caps notify plugs aacparse-2, and so on until the cgo stack // overflows and the process crashes. Stopping autoplug before any // parser is added keeps that loop from ever starting. autoplugID, err := parsebin.Connect("autoplug-continue", func(_ *gst.Element, _ *gst.Pad, caps *gst.Caps) bool { if caps == nil { return true } s := caps.GetStructureAt(0) if s == nil { return true } return !elementaryStreamCaps[s.Name()] }) if err != nil { return result, fmt.Errorf("connect autoplug-continue: %w", err) } defer parsebin.HandlerDisconnect(autoplugID)
mp4mux, err := gst.NewElementWithProperties("mp4mux", map[string]any{ "name": "vod-mp4mux", "fragment-mode": 0, "fragment-duration": 1, }) if err != nil { return result, fmt.Errorf("create mp4mux: %w", err) } if err := pipeline.Add(mp4mux); err != nil { return result, fmt.Errorf("add mp4mux: %w", err) }
appsink, err := gst.NewElementWithProperties("appsink", map[string]any{ "name": "vod-appsink", "sync": false, "async": false, }) if err != nil { return result, fmt.Errorf("create appsink: %w", err) } if err := pipeline.Add(appsink); err != nil { return result, fmt.Errorf("add appsink: %w", err) } if err := mp4mux.Link(appsink); err != nil { return result, fmt.Errorf("link mp4mux -> appsink: %w", err) }
sink := app.SinkFromElement(appsink) sink.SetCallbacks(&app.SinkCallbacks{ NewSampleFunc: WriterNewSample(ctx, prog), })
// padErr captures the first wiring error from pad-added so we can // surface it after the bus loop returns. gotVideo/gotAudio prevent us // from blindly wiring multiple streams of the same kind into mp4mux // (we'd just get the first one; warn the user about extras). // videoCaps/audioCaps capture the actual codec strings so we can // attach them to the trace span. var ( mu sync.Mutex padErr error gotVideo bool gotAudio bool videoCaps string audioCaps string )
handlerID, err := parsebin.Connect("pad-added", func(_ *gst.Element, pad *gst.Pad) { mu.Lock() defer mu.Unlock() if padErr != nil { return } caps := padCaps(pad) if caps == nil { log.Warn(ctx, "parsebin pad missing caps; ignoring", "pad", pad.GetName()) return } s := caps.GetStructureAt(0) if s == nil { log.Warn(ctx, "parsebin caps missing structure; ignoring", "pad", pad.GetName()) return } capsName := s.Name() padCtx := log.WithLogValues(ctx, "pad", pad.GetName(), "caps", capsName) log.Debug(padCtx, "parsebin pad-added") switch { case strings.HasPrefix(capsName, "video/"): if gotVideo { log.Warn(padCtx, "additional video pad; ignoring (only one video stream supported)") return } if err := wireVideoChain(padCtx, pipeline, mp4mux, pad, s); err != nil { padErr = err pipeline.Error("video chain failed", err) return } gotVideo = true videoCaps = capsName result.Video = extractVideoTrack(capsName, s) log.Debug(padCtx, "video probe", "codec", result.Video.Codec, "width", result.Video.Width, "height", result.Video.Height, "fps_num", result.Video.FPSNum, "fps_den", result.Video.FPSDen, ) case strings.HasPrefix(capsName, "audio/"): if gotAudio { log.Warn(padCtx, "additional audio pad; ignoring (only one audio stream supported)") return } transcoded, err := wireAudioChain(padCtx, pipeline, mp4mux, pad, s) if err != nil { padErr = err pipeline.Error("audio chain failed", err) return } gotAudio = true audioCaps = capsName // After transcoding (opus/mp3/etc -> AAC), the audio track // in the final fMP4 is AAC. We report what the *output* // looks like, not the input. For passthrough AAC we keep // the caps as-is. result.Audio = extractAudioTrack(capsName, s, transcoded) log.Debug(padCtx, "audio probe", "codec", result.Audio.Codec, "rate", result.Audio.Rate, "channels", result.Audio.Channels, "transcoded", transcoded, ) span.SetAttributes(attribute.Bool("audio_transcode", transcoded)) default: log.Warn(padCtx, "non-A/V parsebin pad; ignoring") } }) if err != nil { return result, fmt.Errorf("connect pad-added: %w", err) } // Disconnecting before pipeline tear-down breaks the closure->pipeline // reference cycle that would otherwise pin every element alive past // the function's return. defer parsebin.HandlerDisconnect(handlerID)
busErr := make(chan error, 1) go func() { busErr <- HandleBusMessages(ctx, pipeline) }()
if err := pipeline.SetState(gst.StatePlaying); err != nil { span.RecordError(err) return result, fmt.Errorf("set playing: %w", err) } log.Debug(ctx, "pipeline transitioned to PLAYING")
// Watchdog + heartbeat. The bus loop polls ctx.Err() every ~1s, so a // cancel here unblocks the <-busErr wait below; without it a demuxer // that reaches PLAYING but never emits output hangs this goroutine // indefinitely. The same ticker logs an info-level progress line so a // healthy run is visible and a stall ("output_bytes=0" repeating) is // obvious before the timeout even fires. prog.resetTimer() watchdogDone := make(chan struct{}) go func() { ticker := time.NewTicker(vodProgressInterval) defer ticker.Stop() for { select { case <-watchdogDone: return case <-ctx.Done(): return case <-ticker.C: idle := prog.idle() bytes, samples := prog.bytes.Load(), prog.samples.Load() if idle >= vodStallTimeout { prog.stalled.Store(true) log.Error(ctx, "vod pipeline stalled; cancelling", "output_bytes", bytes, "samples", samples, "idle_seconds", idle.Seconds()) cancel() return } log.Log(ctx, "vod pipeline progress", "output_bytes", bytes, "samples", samples, "idle_seconds", idle.Seconds()) } } }()
busErrVal := <-busErr close(watchdogDone) if busErrVal != nil { if prog.stalled.Load() { err := fmt.Errorf("%w: no output for %s (%d bytes in %d samples)", ErrPipelineStalled, vodStallTimeout, prog.bytes.Load(), prog.samples.Load()) span.RecordError(err) span.SetStatus(codes.Error, "stalled") return result, err } mu.Lock() pe := padErr mu.Unlock() if pe != nil { span.RecordError(pe) span.SetStatus(codes.Error, "pad-wiring") return result, pe } span.RecordError(busErrVal) span.SetStatus(codes.Error, "bus") return result, fmt.Errorf("pipeline: %w", busErrVal) } log.Debug(ctx, "pipeline EOS reached cleanly")
// Query the pipeline's duration before tearing it down. gstreamer // reports it in nanoseconds (FormatTime); we expose milliseconds // because that's what place.stream.video's duration field takes. if ok, durNS := pipeline.QueryDuration(gst.FormatTime); ok { result.DurationMS = durNS / int64(time.Millisecond) span.SetAttributes(attribute.Int64("duration_ms", result.DurationMS)) log.Debug(ctx, "queried pipeline duration", "duration_ms", result.DurationMS) } else { log.Warn(ctx, "pipeline duration query failed; record will report 0 ms") }
mu.Lock() defer mu.Unlock() span.SetAttributes( attribute.Bool("got_video", gotVideo), attribute.Bool("got_audio", gotAudio), attribute.String("video_caps", videoCaps), attribute.String("audio_caps", audioCaps), ) if padErr != nil { span.RecordError(padErr) span.SetStatus(codes.Error, "pad-wiring") return result, padErr } if !gotVideo { err := fmt.Errorf("no video stream found in input") span.RecordError(err) span.SetStatus(codes.Error, "no_video") return result, err } return result, nil}
// extractVideoTrack pulls codec/dimensions/framerate out of a parsebin// video pad's caps Structure. The framerate is a GstFraction that// go-gst doesn't expose typed access for, so we stringify and parse// num/den ourselves — same workaround the existing media_data_parser// uses for live-segment probing.func extractVideoTrack(capsName string, s *gst.Structure) *VODVideoTrack { t := &VODVideoTrack{Codec: strings.TrimPrefix(capsName, "video/")} if v, _ := s.GetValue("width"); v != nil { if w, ok := v.(int); ok { t.Width = w } } if v, _ := s.GetValue("height"); v != nil { if h, ok := v.(int); ok { t.Height = h } } if v, _ := s.GetValue("framerate"); v != nil { t.FPSNum, t.FPSDen = parseFraction(v) } return t}
// extractAudioTrack pulls codec/rate/channels out of a parsebin audio// pad's caps Structure. When the chain transcodes to AAC (the common// case for anything that isn't already AAC), the reported codec is// "mpeg" with mpegversion=4, since that's the actual MP4 output —// callers downstream care about what landed in the final blob, not// what came in.func extractAudioTrack(capsName string, s *gst.Structure, transcoded bool) *VODAudioTrack { t := &VODAudioTrack{Codec: strings.TrimPrefix(capsName, "audio/")} if v, _ := s.GetValue("rate"); v != nil { if r, ok := v.(int); ok { t.Rate = r } } if v, _ := s.GetValue("channels"); v != nil { if c, ok := v.(int); ok { t.Channels = c } } if v, _ := s.GetValue("mpegversion"); v != nil { if mv, ok := v.(int); ok { t.MPEGVersion = mv } } if transcoded { // The chain re-encoded to AAC; the final fMP4 carries AAC // regardless of what came in. t.Codec = "mpeg" t.MPEGVersion = 4 } return t}
// parseFraction takes a GstFraction-shaped value from a caps Structure// (go-gst returns it as a stringer with "num/den" format) and returns// the numerator and denominator. Returns 0,0 on any parse failure.func parseFraction(v any) (num, den int) { str := fmt.Sprintf("%v", v) parts := strings.SplitN(str, "/", 2) if len(parts) != 2 { return 0, 0 } n, err1 := strconv.Atoi(strings.TrimSpace(parts[0])) d, err2 := strconv.Atoi(strings.TrimSpace(parts[1])) if err1 != nil || err2 != nil { return 0, 0 } return n, d}
type elemSpec struct { name string props map[string]any}
func wireVideoChain(ctx context.Context, pipeline *gst.Pipeline, mp4mux *gst.Element, pad *gst.Pad, caps *gst.Structure) error { codec := caps.Name() if codec != "video/x-h264" { return fmt.Errorf("%w: video codec %q (only h264 is supported)", ErrUnsupportedCodec, codec) } chain, err := buildChain(pipeline, "vod-video", []elemSpec{ {name: "queue"}, {name: "h264parse", props: map[string]any{"config-interval": 0, "disable-passthrough": true}}, }) if err != nil { return err } if err := linkChain(chain); err != nil { return err } sinkPad := mp4mux.GetRequestPad("video_%u") if sinkPad == nil { return fmt.Errorf("mp4mux: no video request pad") } if ret := pad.Link(chain[0].GetStaticPad("sink")); ret != gst.PadLinkOK { return fmt.Errorf("link parsebin -> video chain: %v", ret) } if ret := chain[len(chain)-1].GetStaticPad("src").Link(sinkPad); ret != gst.PadLinkOK { return fmt.Errorf("link video chain -> mp4mux: %v", ret) } syncAll(chain) log.Debug(ctx, "wired video chain", "codec", codec, "elements", chainNames(chain)) return nil}
// wireAudioChain returns whether the chain inserted a transcode step// (i.e., the input wasn't already AAC).func wireAudioChain(ctx context.Context, pipeline *gst.Pipeline, mp4mux *gst.Element, pad *gst.Pad, caps *gst.Structure) (transcoded bool, err error) { codec := caps.Name() specs, transcoded, err := audioChainSpec(codec, caps) if err != nil { return false, err } chain, err := buildChain(pipeline, "vod-audio", specs) if err != nil { return false, err } if err := linkChain(chain); err != nil { return false, err } sinkPad := mp4mux.GetRequestPad("audio_%u") if sinkPad == nil { return false, fmt.Errorf("mp4mux: no audio request pad") } if ret := pad.Link(chain[0].GetStaticPad("sink")); ret != gst.PadLinkOK { return false, fmt.Errorf("link parsebin -> audio chain: %v", ret) } if ret := chain[len(chain)-1].GetStaticPad("src").Link(sinkPad); ret != gst.PadLinkOK { return false, fmt.Errorf("link audio chain -> mp4mux: %v", ret) } syncAll(chain) log.Debug(ctx, "wired audio chain", "codec", codec, "transcoded", transcoded, "elements", chainNames(chain)) return transcoded, nil}
func chainNames(elems []*gst.Element) []string { names := make([]string, len(elems)) for i, e := range elems { names[i] = e.GetName() } return names}
// transcodeToAAC is the suffix used after a decoder when re-encoding to// AAC for ingest into mp4mux. resample + convert insulate the encoder// from arbitrary sample rates and channel layouts.var transcodeToAAC = []elemSpec{ {name: "audioconvert"}, {name: "audioresample"}, {name: "fdkaacenc"}, {name: "aacparse"},}
// audioChainSpec returns the per-element chain needed to land the given// codec on mp4mux's audio sink, plus a flag indicating whether the chain// re-encodes (i.e., input != AAC). Only the codecs whose decoders/encoders// are actually linked into the static build are supported here: AAC// (aacparse passthrough) and Opus (opusdec → fdkaacenc). MP3, Vorbis,// FLAC, etc. are rejected with ErrUnsupportedCodec — the decoders they'd// need (mpg123audiodec / vorbisdec / flacdec) aren't in the plugin list// in the Makefile.func audioChainSpec(codec string, caps *gst.Structure) (specs []elemSpec, transcoded bool, err error) { out := []elemSpec{{name: "queue"}} switch codec { case "audio/mpeg": v, _ := caps.GetValue("mpegversion") ver, _ := v.(int) if ver != 2 && ver != 4 { return nil, false, fmt.Errorf("%w: audio/mpeg mpegversion=%d", ErrUnsupportedCodec, ver) } // AAC (mpegversion 2 = MPEG-2 AAC, 4 = MPEG-4 AAC); pass through. out = append(out, elemSpec{name: "aacparse"}) return out, false, nil case "audio/x-opus": out = append(out, elemSpec{name: "opusdec"}) out = append(out, transcodeToAAC...) return out, true, nil default: return nil, false, fmt.Errorf("%w: audio codec %q", ErrUnsupportedCodec, codec) }}
func buildChain(pipeline *gst.Pipeline, prefix string, specs []elemSpec) ([]*gst.Element, error) { out := make([]*gst.Element, len(specs)) for i, spec := range specs { props := map[string]any{} for k, v := range spec.props { props[k] = v } props["name"] = fmt.Sprintf("%s-%s-%d", prefix, spec.name, i) e, err := gst.NewElementWithProperties(spec.name, props) if err != nil { return nil, fmt.Errorf("create %s: %w", spec.name, err) } if err := pipeline.Add(e); err != nil { return nil, fmt.Errorf("add %s: %w", spec.name, err) } out[i] = e } return out, nil}
func linkChain(elems []*gst.Element) error { for i := 0; i < len(elems)-1; i++ { if err := elems[i].Link(elems[i+1]); err != nil { return fmt.Errorf("link %s -> %s: %w", elems[i].GetName(), elems[i+1].GetName(), err) } } return nil}
func syncAll(elems []*gst.Element) { for _, e := range elems { e.SyncStateWithParent() }}
// padCaps resolves a pad's current caps with fallbacks for the case// parsebin emits its source pads before the caps event has propagated://// 1. GetCurrentCaps — fast path, works once a CAPS event has fixed// the pad. For many inputs this is set at pad-added time.// 2. GetStream().Caps — parsebin attaches a GstStream to each// exposed pad with the codec caps from its internal parser,// even before the downstream caps query has fixated. Works// reliably for parsebin and decodebin3.// 3. GetAllowedCaps — fallback intersection of the pad's template// and the (still-empty) peer caps. Rarely fixes a codec name// directly but catches typed pad templates as a last resort.//// Returns nil if all three fall short, in which case the caller// should log+ignore the pad — there's no productive way to wire it.func padCaps(pad *gst.Pad) *gst.Caps { if c := pad.GetCurrentCaps(); c != nil { return c } if stream := pad.GetStream(); stream != nil { if c := stream.Caps(); c != nil { return c } } if c := pad.GetAllowedCaps(); c != nil { return c } return nil}