Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
11 kB · 437 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438package cmd
import ( "context" "fmt" "io" "net/http" "strings" "time"
"github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" "github.com/pion/webrtc/v4" pionmedia "github.com/pion/webrtc/v4/pkg/media" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/crypto/spkey" "stream.place/streamplace/pkg/gstinit" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/media")
func WHIP(ctx context.Context, streamKey string, count int, viewers int, duration time.Duration, file string, endpoint string, freezeAfter time.Duration, retry time.Duration) error { if file == "" { return fmt.Errorf("file is required") } if retry < 0 { return fmt.Errorf("retry must not be negative") } gstinit.InitGST()
if duration > 0 { var cancel context.CancelFunc ctx, cancel = context.WithTimeout(ctx, duration) defer cancel() }
w := &WHIPClient{ StreamKey: streamKey, File: file, Endpoint: endpoint, Count: count, FreezeAfter: freezeAfter, Viewers: viewers, }
for { if err := ctx.Err(); err != nil { return err } err := w.WHIP(ctx) if ctx.Err() != nil { return ctx.Err() } if retry == 0 { return err } log.Log(ctx, "WHIP session ended, retrying", "error", err, "delay", retry) select { case <-ctx.Done(): return ctx.Err() case <-time.After(retry): } }}
type WHIPClient struct { StreamKey string File string Endpoint string Count int FreezeAfter time.Duration Viewers int}
var failureStates = []webrtc.ICEConnectionState{ webrtc.ICEConnectionStateFailed, webrtc.ICEConnectionStateDisconnected, webrtc.ICEConnectionStateClosed, webrtc.ICEConnectionStateCompleted,}
type WHIPConnection struct { peerConnection *webrtc.PeerConnection audioTrack *webrtc.TrackLocalStaticSample videoTrack *webrtc.TrackLocalStaticSample did string}
func (w *WHIPClient) WHIP(ctx context.Context) (retErr error) { ctx, cancel := context.WithCancel(ctx) defer cancel()
pipelineSlice := []string{ "filesrc name=filesrc ! qtdemux name=demux", "demux.video_0 ! tee name=video_tee", "demux.audio_0 ! tee name=audio_tee", // sync=true (the default) is load-bearing here, unlike every other // appsink in the tree: these sinks are what pace the file at realtime // so it plays as a live stream — WriteSample pushes to WebRTC // immediately, so without clock sync the whole file would blast // through in one burst. "video_tee. ! queue ! h264parse config-interval=-1 ! video/x-h264,stream-format=byte-stream ! appsink sync=true name=videoappsink", "audio_tee. ! queue ! opusparse ! appsink sync=true name=audioappsink", // "matroskamux name=mux ! fakesink name=fakesink sync=true", // "video_tee. ! mux.video_0", // "audio_tee. ! mux.audio_0", }
pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n")) if err != nil { return err } defer func() { cancel() if err := pipeline.BlockSetState(gst.StateNull); err != nil && retErr == nil { retErr = err } }()
fileSrc, err := pipeline.GetElementByName("filesrc") if err != nil { return err }
if err := fileSrc.Set("location", w.File); err != nil { return err }
videoSink, err := pipeline.GetElementByName("videoappsink") if err != nil { return err }
audioSink, err := pipeline.GetElementByName("audioappsink") if err != nil { return err }
startTime := time.Now() sinks := []*app.Sink{ app.SinkFromElement(videoSink), app.SinkFromElement(audioSink), } // Create accumulators for tracking elapsed duration accumulators := make([]time.Duration, len(sinks))
conns := make([]*WHIPConnection, w.Count) defer func() { for _, conn := range conns { if conn != nil { conn.peerConnection.Close() } } }() g := &errgroup.Group{} for i := 0; i < w.Count; i++ { ctx := ctx // var streamKey string var did string var streamKey string if w.StreamKey != "" { streamKey = w.StreamKey } else { priv, pub, err := spkey.GenerateStreamKey() if err != nil { return err }
did = pub.DIDKey() ctx = log.WithLogValues(ctx, "did", did) streamKey = priv.Multibase() }
g.Go(func() error { conn, err := w.StartWHIPConnection(ctx, streamKey, did) if err != nil { cancel() return err } conns[i] = conn ctx := log.WithLogValues(ctx, "did", did) conn.peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) { log.Log(ctx, "WHIP connection State has changed", "state", connectionState.String()) for _, state := range failureStates { if connectionState == state { log.Log(ctx, "connection failed, cancelling") cancel() } } }) return nil }) }
if err := g.Wait(); err != nil { return err }
// Start a ticker to print elapsed duration every second go func() { ticker := time.NewTicker(time.Second) defer ticker.Stop()
for { select { case <-ctx.Done(): return case <-ticker.C: for i, duration := range accumulators { trackType := "video" if i == 1 { trackType = "audio" } target := startTime.Add(time.Duration(accumulators[i])) diff := time.Since(target) log.Debug(ctx, "elapsed duration", "track", trackType, "duration", duration, "diff", diff) } } } }()
errCh := make(chan error, 1) reportError := func(err error) { select { case errCh <- err: default: } cancel() }
for i := range sinks { func(i int) { sink := sinks[i] trackType := "video" if i == 1 { trackType = "audio" }
sink.SetCallbacks(&app.SinkCallbacks{ NewSampleFunc: func(sink *app.Sink) gst.FlowReturn {
sample := sink.PullSample() if sample == nil { return gst.FlowEOS }
buffer := sample.GetBuffer() if buffer == nil { return gst.FlowError }
samples := buffer.Map(gst.MapRead).Bytes() defer buffer.Unmap()
durationPtr := buffer.Duration().AsDuration() var duration time.Duration if durationPtr == nil { reportError(fmt.Errorf("%v duration: nil", trackType)) return gst.FlowError } else { // fmt.Printf("%v duration: %v\n", trackType, *durationPtr) duration = *durationPtr }
accumulators[i] += duration
if w.FreezeAfter == 0 || time.Since(startTime) < w.FreezeAfter { for _, conn := range conns { if trackType == "video" { if err := conn.videoTrack.WriteSample(pionmedia.Sample{Data: samples, Duration: duration}); err != nil { log.Log(ctx, "error writing video sample", "error", err) reportError(err) return gst.FlowError } } else { if err := conn.audioTrack.WriteSample(pionmedia.Sample{Data: samples, Duration: duration}); err != nil { log.Log(ctx, "error writing video sample", "error", err) reportError(err) return gst.FlowError } } } }
return gst.FlowOK }, }) }(i) }
go func() { if err := media.HandleBusMessages(ctx, pipeline); err != nil { log.Log(ctx, "pipeline error", "error", err) } cancel() }()
if err = pipeline.SetState(gst.StatePlaying); err != nil { return err } if w.Viewers > 0 { whepG, ctx := errgroup.WithContext(ctx) for i := 0; i < w.Count; i++ { did := conns[i].did w := &WHEPClient{ Endpoint: fmt.Sprintf("%s/api/playback/%s/webrtc", w.Endpoint, did), Count: w.Viewers, } whepG.Go(func() error { return w.WHEP(ctx) }) } if err := whepG.Wait(); err != nil { return err } }
<-ctx.Done()
// reportError cancels the session context right after queueing the error it // saw, so both branches below would be ready at once and the select would // pick either one. Prefer the queued error: it is the actual failure, and // returning context.Canceled instead would hide it from both the one-shot // caller and the retry log. select { case err := <-errCh: return err default: } return ctx.Err()}
func (w *WHIPClient) StartWHIPConnection(ctx context.Context, streamKey string, did string) (*WHIPConnection, error) {
// Prepare the configuration config := webrtc.Configuration{}
// Create a new RTCPeerConnection peerConnection, err := webrtc.NewPeerConnection(config) if err != nil { return nil, err } connected := false defer func() { if !connected { peerConnection.Close() } }()
// Create a audio track audioTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: "audio/opus"}, "audio", "pion1") if err != nil { return nil, err } _, err = peerConnection.AddTrack(audioTrack) if err != nil { return nil, err }
// Create a video track videoTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: "video/h264"}, "video", "pion2") if err != nil { return nil, err } _, err = peerConnection.AddTrack(videoTrack) if err != nil { return nil, err }
// Create an offer offer, err := peerConnection.CreateOffer(nil) if err != nil { return nil, err }
// Set the generated offer as our LocalDescription err = peerConnection.SetLocalDescription(offer) if err != nil { return nil, err }
// Wait for ICE gathering to complete // gatherComplete := webrtc.GatheringCompletePromise(peerConnection) // <-gatherComplete
// Create HTTP client and prepare the request client := &http.Client{}
// Send the WHIP request to the server req, err := http.NewRequestWithContext(ctx, "POST", w.Endpoint, strings.NewReader(offer.SDP)) if err != nil { return nil, err } req.Header.Set("Authorization", "Bearer "+streamKey) req.Header.Set("Content-Type", "application/sdp")
// Execute the request resp, err := client.Do(req) if err != nil { return nil, err } defer resp.Body.Close()
// Read and process the answer answerBytes, err := io.ReadAll(resp.Body) if err != nil { return nil, err }
// Parse the SDP answer var answer webrtc.SessionDescription answer.Type = webrtc.SDPTypeAnswer answer.SDP = string(answerBytes)
// Apply the answer as remote description err = peerConnection.SetRemoteDescription(answer) if err != nil { return nil, err }
gatherComplete := webrtc.GatheringCompletePromise(peerConnection) select { case <-ctx.Done(): return nil, ctx.Err() case <-gatherComplete: }
conn := &WHIPConnection{ peerConnection: peerConnection, audioTrack: audioTrack, videoTrack: videoTrack, did: did, }
connected = true return conn, nil}