Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
10 kB · 323 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324package media
import ( "context" "fmt" "sync" "time"
"github.com/google/uuid" "github.com/pion/webrtc/v4" "github.com/pion/webrtc/v4/pkg/media" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/bus" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/renditions")
// This function remains in scope for the duration of a single users' playbackfunc (mm *MediaManager) WebRTCPlayback2(ctx context.Context, user string, rendition string, offer *webrtc.SessionDescription, viewer string) (*webrtc.SessionDescription, error) { uu, err := uuid.NewV7() if err != nil { return nil, err } ctx = log.WithLogValues(ctx, "webrtcID", uu.String()) ctx = log.WithLogValues(ctx, "mediafunc", "WebRTCPlayback")
// Create a new RTCPeerConnection peerConnection, err := mm.webrtcAPI.NewPeerConnection(mm.webrtcConfig) if err != nil { return nil, fmt.Errorf("failed to create WebRTC peer connection: %w", err) }
audioOnly := rendition == renditions.AudioRendition.Name
var videoTrack *webrtc.TrackLocalStaticSample var videoRTPSender *webrtc.RTPSender if !audioOnly { videoTrack, err = webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, "video", "pion") if err != nil { return nil, fmt.Errorf("failed to create video track: %w", err) } videoRTPSender, err = peerConnection.AddTrack(videoTrack) if err != nil { return nil, fmt.Errorf("failed to add video track to peer connection: %w", err) } }
audioTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus}, "audio", "pion") if err != nil { return nil, fmt.Errorf("failed to create audio track: %w", err) } audioRTPSender, err := peerConnection.AddTrack(audioTrack) if err != nil { return nil, fmt.Errorf("failed to add audio track to peer connection: %w", err) }
close := func() { if cErr := peerConnection.Close(); cErr != nil { log.Log(ctx, "cannot close peerConnection: %v\n", cErr) } }
// Set the remote SessionDescription if err = peerConnection.SetRemoteDescription(*offer); err != nil { close() return nil, fmt.Errorf("failed to set remote description: %w", err) }
// Create answer answer, err := peerConnection.CreateAnswer(nil) if err != nil { close() return nil, fmt.Errorf("failed to create answer: %w", err) }
// Sets the LocalDescription, and starts our UDP listeners if err = peerConnection.SetLocalDescription(answer); err != nil { close() return nil, fmt.Errorf("failed to set local description: %w", err) }
// Create channel that is blocked until ICE Gathering is complete gatherComplete := webrtc.GatheringCompletePromise(peerConnection)
// Setup complete! Now we boot up streaming in the background while returning the SDP offer to the user.
// The session only counts as a viewer once the peer connection actually // establishes — counting at SDP-answer time inflated the count with // handshakes that never connected (each lingering until ICE failure // detection). The mutex pairs the increment with exactly one decrement // even if a connect races session teardown. var viewerMu sync.Mutex viewerCounted := false viewerDone := false markConnected := func() { viewerMu.Lock() defer viewerMu.Unlock() if viewerDone || viewerCounted { return } viewerCounted = true mm.IncrementViewerCount(user, "webrtc") } markDone := func() { viewerMu.Lock() defer viewerMu.Unlock() viewerDone = true if viewerCounted { viewerCounted = false mm.DecrementViewerCount(user, "webrtc") } }
go func() { ctx, cancel := context.WithCancel(ctx) defer cancel() defer markDone()
latency := time.Duration(0)
packetQueue := make(chan *bus.PacketizedSegment, 1024) go func() { busRendition := rendition if audioOnly { busRendition = "source" } segChan := mm.bus.SubscribeSegmentBuf(ctx, user, busRendition, 2) defer mm.bus.UnsubscribeSegment(ctx, user, busRendition, segChan) for { select { case <-ctx.Done(): log.Debug(ctx, "exiting segment reader") return case file := <-segChan.C: log.Debug(ctx, "got segment", "file", file.Filepath) if !file.Published && viewer != user && !mm.cli.WideOpen { log.Warn(ctx, "segment is not published and viewer is not the user", "viewer", viewer, "user", user) continue } latency += file.PacketizedData.Duration packetQueue <- file.PacketizedData } } }()
go func() { go func() { <-ctx.Done() if cErr := peerConnection.Close(); cErr != nil { log.Log(ctx, "cannot close peerConnection: %v\n", cErr) } }()
var scalar float64 = 1
for { select { case <-ctx.Done(): return case packet := <-packetQueue: latency -= packet.Duration scalar = getPlaybackRate(latency) log.Debug(ctx, "playback latency", "latency", latency, "scalar", scalar) g, _ := errgroup.WithContext(ctx) wroteAny := false
if !audioOnly && len(packet.Video) > 0 { wroteAny = true g.Go(func() error { return writeSamples(ctx, videoTrack, packet.Video, scalar) }) } else if !audioOnly { log.Warn(ctx, "no video samples to write") } // The audio path deliberately keeps the original // uniform-duration ticker instead of writeSamples: Opus // packets are a constant 20ms, so the uniform split is // exact, and this path is proven in production — audio // timing regressions are immediately audible as garbling. // Video is the track that needs per-sample durations (its // spacing goes non-uniform when an encoder sheds frames). var audioDur time.Duration if len(packet.Audio) > 0 { audioDur = packet.Duration / time.Duration(len(packet.Audio)) } if audioDur > 0 { wroteAny = true g.Go(func() error { ticker := time.NewTicker(time.Duration(float64(audioDur) * (1 / scalar))) defer ticker.Stop() for _, audio := range packet.Audio { err := audioTrack.WriteSample(media.Sample{Data: audio.Data, Duration: audioDur}) if err != nil { return fmt.Errorf("failed to write audio sample: %w", err) } select { case <-ctx.Done(): return nil case <-ticker.C: continue } } return nil }) } else { log.Warn(ctx, "no audio samples to write") } if wroteAny { if err := g.Wait(); err != nil { log.Error(ctx, "failed to write samples", "error", err) cancel() } } } } }()
if !audioOnly { go func() { rtcpBuf := make([]byte, 1500) for { if _, _, rtcpErr := videoRTPSender.Read(rtcpBuf); rtcpErr != nil { return } } }() }
go func() { rtcpBuf := make([]byte, 1500) for { if _, _, rtcpErr := audioRTPSender.Read(rtcpBuf); rtcpErr != nil { return } } }()
// Set the handler for ICE connection state // This will notify you when the peer has connected/disconnected peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) { log.Log(ctx, "Connection State has changed", "state", connectionState.String()) })
// Set the handler for Peer connection state // This will notify you when the peer has connected/disconnected peerConnection.OnConnectionStateChange(func(s webrtc.PeerConnectionState) { log.Log(ctx, "Peer Connection State has changed", "state", s.String())
if s == webrtc.PeerConnectionStateConnected { markConnected() }
if s == webrtc.PeerConnectionStateFailed || s == webrtc.PeerConnectionStateClosed || s == webrtc.PeerConnectionStateDisconnected { // Wait until PeerConnection has had no network activity for 30 seconds or another failure. It may be reconnected using an ICE Restart. // Use webrtc.PeerConnectionStateDisconnected if you are interested in detecting faster timeout. // Note that the PeerConnection may come back from PeerConnectionStateDisconnected. log.Log(ctx, "Peer Connection has gone to failed, exiting") cancel() } })
<-ctx.Done()
log.Warn(ctx, "exiting playback")
}() select { case <-gatherComplete: return peerConnection.LocalDescription(), nil case <-ctx.Done(): return nil, ctx.Err() }}
// writeSamples writes one track's samples paced by their real durations —// the source timeline, so non-uniform frame spacing (bursts and gaps from an// encoder shedding frames) reaches the viewer intact. The stamped duration// stays real even while catch-up (scalar > 1) speeds the pacing: the// receiver's clock is authoritative for playout, sending faster just refills// its buffer.//// Pacing is against a wall-clock schedule, NOT a per-sample sleep: each sleep// overshoots by scheduler latency, and chaining sleeps accumulates that// overshoot into real drift (hundreds of ms per segment at high sample rates// — enough to starve the receiver's jitter buffer). Sleeping until the// scheduled deadline instead absorbs overshoot in the next iteration, like a// ticker does.func writeSamples(ctx context.Context, track *webrtc.TrackLocalStaticSample, samples []bus.PacketizedSample, scalar float64) error { start := time.Now() var scheduled time.Duration for _, s := range samples { if err := track.WriteSample(media.Sample{Data: s.Data, Duration: s.Duration}); err != nil { return fmt.Errorf("failed to write sample: %w", err) } scheduled += time.Duration(float64(s.Duration) / scalar) wait := scheduled - time.Since(start) if wait <= 0 { continue } timer := time.NewTimer(wait) select { case <-ctx.Done(): timer.Stop() return nil case <-timer.C: } } return nil}
// getPlaybackRate returns a playback rate that eases from 1.0 to 1.5 between 7 and 60 secondsfunc getPlaybackRate(dur time.Duration) float64 { switch { case dur <= 7*time.Second: return 1.0 case dur >= 60*time.Second: return 1.5 default: // Linear interpolation between (7,1.0) and (60,1.5) progress := (float64(dur) - float64(7*time.Second)) / (float64(60*time.Second) - float64(7*time.Second)) return 1.0 + (0.5 * progress) }}