Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
6.4 kB · 165 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166package media
import ( "bytes" "context" "fmt" "io" "sort"
"github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/muxl")
// SignSegmentStreamFunc drives muxl-sign's streaming per-segment signer over an// fMP4 input, emitting one signed-segment event per GoP on eventCh. It is the// only thing muxlSignSegmentElem needs from a signer, so the isolated ingest// worker can supply a key-PEM-backed closure without a full MediaSigner (and// without the model/DB a MediaSignerLocal carries).type SignSegmentStreamFunc func(ctx context.Context, input io.Reader, eventCh chan *muxl.MuxlEvent) error
// MuxlSignSegmentElem builds the gstreamer bin that muxes the incoming// video+audio into a fragmented MP4 stream, then drives muxl-sign's streaming// per-segment signer over it. For each GoP it assembles the bare canonical// .m4s — the per-track signed [c2pa-uuid][muxl-uuid][moof][mdat] runs// concatenated in track-id order — and hands it to onSegment. That bare .m4s// is exactly what gets stored, verified, and replicated; no flat MP4 is// produced here. Presentation headers are synthesized downstream (ValidateMP4// / playback) only when needed.func MuxlSignSegmentElem(ctx context.Context, cli *config.CLI, ms MediaSigner, onSegment func(ctx context.Context, segment []byte) error) (*gst.Element, error) { elem, _, err := muxlSignSegmentElem(ctx, cli, ms.SignSegmentStream, onSegment) return elem, err}
// muxlSignSegmentElem is MuxlSignSegmentElem's core, parameterized by the raw// sign-stream function and additionally returning a done channel that closes// once every signed segment has been drained to onSegment (the signer goroutine// has finished and the event loop has emptied). The isolated ingest worker waits// on it to guarantee all segment frames are flushed before it signals a clean// end-of-stream.func muxlSignSegmentElem(ctx context.Context, cli *config.CLI, signStream SignSegmentStreamFunc, onSegment func(ctx context.Context, segment []byte) error) (*gst.Element, <-chan struct{}, error) { ctx = log.WithLogValues(ctx, "func", "MuxlSignSegmentElem") bin := gst.NewBin("muxl-segment-bin") elem, err := gst.NewElementWithProperties("mp4mux", map[string]any{ "name": "fmp4mux", "fragment-mode": 0, "fragment-duration": 1, }) if err != nil { return nil, nil, err } if err := bin.Add(elem); err != nil { return nil, nil, fmt.Errorf("failed to add mp4mux to bin: %w", err) }
videoPad := elem.GetRequestPad("video_%u") if videoPad == nil { return nil, nil, fmt.Errorf("failed to get video pad") } videoGhost := gst.NewGhostPad("video_0", videoPad) if videoGhost == nil { return nil, nil, fmt.Errorf("failed to create video ghost pad") } audioPad := elem.GetRequestPad("audio_%u") if audioPad == nil { return nil, nil, fmt.Errorf("failed to get audio pad") } audioGhost := gst.NewGhostPad("audio_0", audioPad) if audioGhost == nil { return nil, nil, fmt.Errorf("failed to create audio ghost pad") } if ok := bin.AddPad(videoGhost.Pad); !ok { return nil, nil, fmt.Errorf("failed to add video ghost pad to bin") } if ok := bin.AddPad(audioGhost.Pad); !ok { return nil, nil, fmt.Errorf("failed to add audio ghost pad to bin") }
// sync=false: this sink feeds the signer, not a display — render as fast as // upstream produces. The default (sync=true) made the appsink wait on the // pipeline clock per buffer, pacing the whole ingest graph at realtime: // harmless for a live source arriving at 1x, but it throttled tests/replays // and kept the graph's queues near-full for no benefit. Every other appsink // in the tree already sets this. appsink, err := gst.NewElementWithProperties("appsink", map[string]any{ "name": "muxl-appsink", "sync": false, }) if err != nil { return nil, nil, fmt.Errorf("failed to create appsink element: %w", err) } if err := bin.Add(appsink); err != nil { return nil, nil, fmt.Errorf("failed to add appsink to bin: %w", err) } if err := elem.Link(appsink); err != nil { return nil, nil, fmt.Errorf("failed to link mp4mux to appsink: %w", err) }
r, w := io.Pipe() go func() { <-ctx.Done() r.Close() }()
// The signer and its event drain run on a non-cancellable ctx: cancelling // ctx is the FLUSH signal, not an abort — it closes the input pipe above, // the signer sees EOF, signs the final GoP, and exits cleanly. If the // cancelled ctx reached muxl's event parser instead, the parser would // abandon the stream mid-write and the signer wasm would deadlock against // the unread stdout pipe — done would never close and the caller's drain // (`cancel(); <-done`) would hang forever. That was rare while ingest was // clock-paced (everything had drained by EOS); at full speed EOS+cancel // land while GoPs are still in flight, and the abort path lost every time. drainCtx := context.WithoutCancel(ctx)
// Stream the fMP4 through the per-segment signer; each event carries one // GoP's per-track signed canonical segments. eventCh := make(chan *muxl.MuxlEvent, 16) go func() { err := signStream(drainCtx, r, eventCh) close(eventCh) if err != nil && ctx.Err() == nil { log.Error(ctx, "error running muxl sign-segment", "error", err) } }() done := make(chan struct{}) go func() { defer close(done) for ev := range eventCh { if ev.Type != "signed-segment" { continue } segment := concatTracksSorted(ev.Tracks) cli.DumpDebugSegment(drainCtx, "muxl_signed_segment.m4s", bytes.NewReader(segment)) if err := onSegment(drainCtx, segment); err != nil { log.Error(drainCtx, "error handling signed segment", "error", err) } } }()
sink := app.SinkFromElement(appsink) sink.SetCallbacks(&app.SinkCallbacks{ NewSampleFunc: WriterNewSample(ctx, w), })
return bin.Element, done, nil}
// concatTracksSorted joins the per-track canonical segment bytes for one GoP// in ascending track-id order — the canonical interleave a multi-track .m4s// uses, which muxl's unwrap/verify/wrap all expect.func concatTracksSorted(tracks map[string][]byte) []byte { keys := make([]string, 0, len(tracks)) for k := range tracks { keys = append(keys, k) } sort.Strings(keys) var out []byte for _, k := range keys { out = append(out, tracks[k]...) } return out}