Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
12 kB · 375 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376package media
import ( "bytes" "context" "crypto/tls" "fmt" "io" "net" "net/url" "strings" "time"
"github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/muxl" "stream.place/streamplace/pkg/placestream")
// RTMPPush is the in-process multistream egress: it assembles the streamer's// source segments into a continuous fMP4 stream and runs the native RTMP push// pipeline over it, reporting status straight to the DB. The isolated// counterpart (RTMPPushIsolated) runs the same native pipeline in a worker// subprocess so a gst fault in the egress chain can't take the node down.func (mm *MediaManager) RTMPPush(ctx context.Context, user string, rendition string, targetView *placestream.MultistreamDefs_TargetView) error { ctx, cancel := context.WithCancel(ctx) defer cancel() ctx = log.WithLogValues(ctx, "mediafunc", "RTMPPush") rec, ok := targetView.Record.Val.(*placestream.MultistreamTarget) if !ok { return fmt.Errorf("failed to convert target view to multistream target") }
// Source: subscribe to the streamer's segments and assemble one continuous // fMP4 stream for the push pipeline. Tied to ctx so it tears down when the // pipeline returns. pr, pw := io.Pipe() go func() { pw.CloseWithError(mm.writeRTMPSource(ctx, user, rendition, pw)) }()
// Status straight to the DB (in-process). The isolated worker reports the // same events back over a frame channel instead, where the supervisor writes // them — see runRTMPPushPipeline / RTMPPushIsolated. report := func(status, message string) { if err := mm.atsync.StatefulDB.CreateMultistreamEvent(targetView.Uri, message, status); err != nil { log.Error(ctx, "failed to create multistream event", "error", err) } } return mm.runRTMPPushPipeline(ctx, pr, rec.Url, report)}
// writeRTMPSource subscribes to the streamer's source segments, selects the AAC// audio + video tracks from each dual-codec segment, synthesizes a single fMP4// init from the first segment, and writes one continuous fMP4 stream to w (init// then every segment's canonical bytes concatenated). It returns when ctx is// done or a select/encode/write fails; the caller owns closing w.//// MUXL segments carry per-track monotonic tfdt, so blind concatenation after a// single synthesized init is a valid fMP4 timeline with no remux. The init// reflects the first segment's catalog and is never re-emitted; muxl derives the// catalog from the moov and does not parse the H.264 bitstream, so a mid-stream// resolution/orientation change (carried in-band as new SPS/PPS at a keyframe)// is invisible to it — the parameter sets pass through verbatim to// h264parse/flvmux and the init's declared dimensions simply stay at the initial// config. Reflecting such a change in container metadata would require parsing// SPS/PPS.func (mm *MediaManager) writeRTMPSource(ctx context.Context, user, rendition string, w io.Writer) error { segChan := mm.bus.SubscribeSegment(ctx, user, rendition) defer mm.bus.UnsubscribeSegment(ctx, user, rendition, segChan) first := true for { select { case <-ctx.Done(): return ctx.Err() case seg := <-segChan.C: log.Debug(ctx, "segment received", "file", seg.Filepath) if len(seg.Muxl) == 0 { log.Warn(ctx, "source segment has no MUXL bytes, skipping", "file", seg.Filepath) continue } // RTMP wants AAC: select video + the AAC audio track from the // dual-codec segment and feed only those, so flvmux gets AAC with no // transcode. aacSeg, err := filterSegmentToCodec(ctx, seg.Muxl, false) if err != nil { return fmt.Errorf("select AAC audio: %w", err) } if first { var init bytes.Buffer if err := muxl.RunMuxlWrapInit(ctx, bytes.NewReader(aacSeg), &init); err != nil { return fmt.Errorf("synthesize init segment: %w", err) } log.Debug(ctx, "init segment synthesized", "size", init.Len()) if _, err := w.Write(init.Bytes()); err != nil { return err } first = false } log.Debug(ctx, "writing segment", "size", len(aacSeg)) if _, err := w.Write(aacSeg); err != nil { return err } } }}
// runRTMPPushPipeline builds and runs the native RTMP egress pipeline:// appsrc(source) → qtdemux → {h264parse, aacparse} → flvmux → rtmp2sink → a// local TCP/TLS forwarder → the target URL. Status updates (currently "active"// once the server acks bytes) go through report. This is the crash-prone native// core shared by the in-process RTMPPush and the isolated rtmp-push worker; only// `source` and `report` differ between them, so the two paths run an identical// gst pipeline and can't drift.func (mm *MediaManager) runRTMPPushPipeline(ctx context.Context, source io.Reader, targetURL string, report func(status, message string)) error { ctx, cancel := context.WithCancel(ctx) defer cancel()
pipelineSlice := []string{ "appsrc name=muxlsrc ! qtdemux name=demux", "flvmux name=muxer ! rtmp2sink name=rtmp2sink", fmt.Sprintf("%s name=videoqueue ! h264parse ! muxer.video", constants.Queue2Big), // Segments carry AAC (we feed only the AAC track), so pass it straight to // flvmux — no Opus→AAC transcode. fmt.Sprintf("%s name=audioqueue ! aacparse ! muxer.audio", constants.Queue2Big), }
pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n")) if err != nil { return fmt.Errorf("failed to create GStreamer pipeline: %w", err) //nolint:all }
rtmp2sink, err := pipeline.GetElementByName("rtmp2sink") if err != nil { return fmt.Errorf("failed to get rtmp2sink element from pipeline: %w", err) }
// rtmp2sink can't speak rtmps and doesn't do TLS, so we always point it at a // localhost forwarder that relays (optionally over TLS) to the real target. u, err := url.Parse(targetURL) if err != nil { return fmt.Errorf("failed to parse target URL: %w", err) } var localAddr string switch u.Scheme { case "rtmps": localAddr, err = mm.RunTLSFForwarder(ctx, targetURL) case "rtmp": localAddr, err = mm.RunTCPForwarder(ctx, targetURL) default: return fmt.Errorf("invalid target URL scheme: %s", u.Scheme) } if err != nil { return fmt.Errorf("failed to run forwarder: %w", err) } local := fmt.Sprintf("rtmp://%s%s", localAddr, u.Path) log.Debug(ctx, "running forwarder", "localAddr", local) if err := rtmp2sink.SetProperty("location", local); err != nil { return fmt.Errorf("failed to set rtmp2sink location: %w", err) }
// Poll the sink and report "active" once the destination acks bytes; back off // once it's flowing so we don't spam the status channel. go func() { pollFreq := time.Second * 1 for { select { case <-ctx.Done(): return case <-time.After(pollFreq): prop, err := rtmp2sink.GetProperty("stats") if err != nil { log.Error(ctx, "error getting rtmp2sink stats", "error", err) continue } if prop == nil { log.Error(ctx, "failed to get rtmp2sink stats", "prop", prop) continue } propVal, ok := prop.(*gst.Structure) if !ok { log.Error(ctx, "failed to convert rtmp2sink stats", "prop", prop) continue } outBytesAcked, err := propVal.GetValue("out-bytes-acked") if err != nil { log.Error(ctx, "failed to get rtmp2sink out-bytes-acked", "error", err) continue } outBytesAckedVal, ok := outBytesAcked.(uint64) if !ok { log.Error(ctx, "failed to convert rtmp2sink out-bytes-acked", "prop", prop) continue } if outBytesAckedVal > 0 { report("active", fmt.Sprintf("wrote %d bytes", outBytesAckedVal)) // once it's working we don't need to spam the status channel pollFreq = time.Second * 15 } log.Debug(ctx, "rtmp2sink out-bytes-acked", "outBytesAckedVal", outBytesAckedVal) } } }()
muxlSrc, err := pipeline.GetElementByName("muxlsrc") if err != nil { return fmt.Errorf("failed to get appsrc element from pipeline: %w", err) } app.SrcFromElement(muxlSrc).SetCallbacks(&app.SourceCallbacks{ NeedDataFunc: ReaderNeedDataIncremental(ctx, source), })
videoQueue, err := pipeline.GetElementByName("videoqueue") if err != nil { return fmt.Errorf("failed to get video parse element from pipeline: %w", err) } audioQueue, err := pipeline.GetElementByName("audioqueue") if err != nil { return fmt.Errorf("failed to get audio parse element from pipeline: %w", err) }
// qtdemux exposes its track pads only after parsing the moov, so link them // on pad-added: video → h264parse, audio → aacparse. demux, err := pipeline.GetElementByName("demux") if err != nil { return fmt.Errorf("failed to get demux element from pipeline: %w", err) } if _, err := demux.Connect("pad-added", func(self *gst.Element, pad *gst.Pad) { name := pad.GetName() var sink *gst.Pad switch { case strings.HasPrefix(name, "video_"): sink = videoQueue.GetStaticPad("sink") case strings.HasPrefix(name, "audio_"): sink = audioQueue.GetStaticPad("sink") default: log.Debug(ctx, "ignoring demux pad", "name", name) return } if linked := pad.Link(sink); linked != gst.PadLinkOK { log.Error(ctx, "failed to link demux pad", "name", name, "result", linked) } }); err != nil { return fmt.Errorf("failed to connect demux pad-added: %w", err) }
errCh := make(chan error) go func() { err := HandleBusMessages(ctx, pipeline) log.Log(ctx, "RTMP push pipeline error", "error", err) errCh <- err }()
err = pipeline.SetState(gst.StatePlaying) if err != nil { return fmt.Errorf("failed to set pipeline state to playing: %w", err) }
defer func() { log.Log(ctx, "shutting down RTMP push pipeline") if err := pipeline.SetState(gst.StateNull); err != nil { log.Error(ctx, "failed to set pipeline state to null", "error", err) } }()
return <-errCh}
func (mm *MediaManager) RunTLSFForwarder(ctx context.Context, dest string) (string, error) { destURL, err := url.Parse(dest) if err != nil { return "", fmt.Errorf("failed to parse destination URL: %w", err) } return mm.runForwarder(ctx, dest, func(destHost string) (net.Conn, error) { return tls.Dial("tcp", destHost, &tls.Config{ ServerName: destURL.Hostname(), }) })}
func (mm *MediaManager) RunTCPForwarder(ctx context.Context, dest string) (string, error) { return mm.runForwarder(ctx, dest, func(destHost string) (net.Conn, error) { return net.Dial("tcp", destHost) })}
func (mm *MediaManager) runForwarder(ctx context.Context, dest string, dial func(destHost string) (net.Conn, error)) (string, error) { ctx = log.WithLogValues(ctx, "mediafunc", "runForwarder") // Parse the destination URL to extract host and port destURL, err := url.Parse(dest) if err != nil { return "", fmt.Errorf("failed to parse destination URL: %w", err) }
// Default to port 1935 if not specified destHost := destURL.Host if !strings.Contains(destHost, ":") { destHost = destHost + ":1935" }
// Listen on a random port listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { return "", fmt.Errorf("failed to listen on random port: %w", err) }
log.Debug(ctx, "RTMP forwarder listening", "localAddr", listener.Addr().String(), "destination", dest)
go func() { <-ctx.Done() listener.Close() }()
go func() { defer listener.Close() if ctx.Err() != nil { return } // Accept incoming RTMP connection clientConn, err := listener.Accept() if err != nil { log.Error(ctx, "failed to accept connection", "error", err) return }
closed := false go func() { <-ctx.Done() if !closed { closed = true clientConn.Close() } }()
defer func() { if !closed { closed = true clientConn.Close() } }()
// Establish connection to destination serverConn, err := dial(destHost) if err != nil { log.Error(ctx, "failed to establish connection to destination", "error", err) return } defer serverConn.Close()
// Proxy data bidirectionally done := make(chan error, 2)
// Copy from client to server go func() { _, err := io.Copy(serverConn, clientConn) done <- err }()
// Copy from server to client go func() { _, err := io.Copy(clientConn, serverConn) done <- err }()
// Wait for either direction to complete or error err = <-done if err != nil { log.Error(ctx, "proxy connection error", "error", err) } }()
return listener.Addr().String(), nil}