Something went wrong. Try again.
Something went wrong. Try again.
9.5 kB · 372 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373package cmd
import ( "context" "fmt" "io" "net/http" "strings" "sync" "time"
"github.com/pion/webrtc/v4" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/log")
func WHEP(ctx context.Context, count int, duration time.Duration, endpoint string) error { if duration > 0 { var cancel context.CancelFunc ctx, cancel = context.WithTimeout(ctx, duration) defer cancel() }
w := &WHEPClient{ Endpoint: endpoint, Count: count, }
return w.WHEP(ctx)}
type WHEPClient struct { StreamKey string File string Endpoint string Count int FreezeAfter time.Duration Stats []map[string]*TrackStats}
type WHEPConnection struct { peerConnection *webrtc.PeerConnection audioTrack *webrtc.TrackLocalStaticSample videoTrack *webrtc.TrackLocalStaticSample did string Done func() <-chan struct{}}
type TrackStats struct { Total int lastTotal int lastUpdate time.Time mu sync.Mutex}
func (w *WHEPClient) StartWHEPConnection(ctx context.Context, stats map[string]*TrackStats) (*WHEPConnection, error) {
// Prepare the configuration config := webrtc.Configuration{}
// Create a new RTCPeerConnection peerConnection, err := webrtc.NewPeerConnection(config) if err != nil { return nil, err }
// Create a ticker to print combined bitrate every 5 seconds ticker := time.NewTicker(5 * time.Second)
// Start a goroutine to print combined bitrate go func() { for { select { case <-ticker.C: currentTime := time.Now()
// Lock both stats to get a consistent snapshot for _, s := range stats { s.mu.Lock() }
videoStats := stats["video"] audioStats := stats["audio"]
videoElapsed := currentTime.Sub(videoStats.lastUpdate).Seconds() audioElapsed := currentTime.Sub(audioStats.lastUpdate).Seconds()
videoBytes := videoStats.Total - videoStats.lastTotal audioBytes := audioStats.Total - audioStats.lastTotal
videoBitrate := float64(videoBytes) * 8 / videoElapsed / 1000 // kbps audioBitrate := float64(audioBytes) * 8 / audioElapsed / 1000 // kbps
log.Log(ctx, "bitrate stats", "video", fmt.Sprintf("%.2f kbps (%.2f KB)", videoBitrate, float64(videoBytes)/1000), "audio", fmt.Sprintf("%.2f kbps (%.2f KB)", audioBitrate, float64(audioBytes)/1000), "total", fmt.Sprintf("%.2f kbps", videoBitrate+audioBitrate))
// Update last values videoStats.lastTotal = videoStats.Total videoStats.lastUpdate = currentTime audioStats.lastTotal = audioStats.Total audioStats.lastUpdate = currentTime
// Unlock stats for _, s := range stats { s.mu.Unlock() }
case <-ctx.Done(): ticker.Stop() return } } }()
go func() { ctx, cancel := context.WithCancel(ctx) peerConnection.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) { log.Log(ctx, "track received", "track", track.ID())
// Determine track type trackType := "video" if track.Kind() == webrtc.RTPCodecTypeAudio { trackType = "audio" }
trackStat := stats[trackType]
for { if ctx.Err() != nil { return } rtp, _, err := track.ReadRTP() if err != nil { log.Log(ctx, "error reading RTP", "error", err) cancel() return }
trackStat.mu.Lock() trackStat.Total += len(rtp.Payload) trackStat.mu.Unlock() } }) peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) { log.Log(ctx, "WHEP connection State has changed", "state", connectionState.String()) for _, state := range failureStates { if connectionState == state { log.Log(ctx, "connection failed, cancelling") cancel() } } })
<-ctx.Done() peerConnection.Close() }() peerConnection.OnICECandidate(func(candidate *webrtc.ICECandidate) { log.Log(ctx, "ICE candidate", "candidate", candidate) }) if _, err := peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, webrtc.RTPTransceiverInit{ Direction: webrtc.RTPTransceiverDirectionRecvonly, }); err != nil { return nil, fmt.Errorf("failed to add video transceiver: %w", err) } if _, err := peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio, webrtc.RTPTransceiverInit{ Direction: webrtc.RTPTransceiverDirectionRecvonly, }); err != nil { return nil, fmt.Errorf("failed to add audio transceiver: %w", 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.NewRequest("POST", w.Endpoint, strings.NewReader(offer.SDP)) if err != nil { return nil, err } 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() // WHEP proper answers 201 Created (the /api/playback/.../webrtc route); // the XRPC method place.stream.playback.whep goes through the generated // handler, which can only write 200. Either carries the SDP answer. if resp.StatusCode != 200 && resp.StatusCode != 201 { body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) return nil, fmt.Errorf("status code: %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) }
// 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) <-gatherComplete
conn := &WHEPConnection{ peerConnection: peerConnection, Done: ctx.Done, }
return conn, nil}
func (w *WHEPClient) WHEP(ctx context.Context) error { w.Stats = []map[string]*TrackStats{} ctx, cancel := context.WithCancel(ctx) defer cancel()
conns := make([]*WHEPConnection, w.Count) g := &errgroup.Group{} for i := 0; i < w.Count; i++ { stats := map[string]*TrackStats{ "video": {lastUpdate: time.Now()}, "audio": {lastUpdate: time.Now()}, } w.Stats = append(w.Stats, stats) g.Go(func() error { conn, err := w.StartWHEPConnection(ctx, stats) if err != nil { return err } conns[i] = conn
<-conn.Done()
return nil }) }
err := g.Wait() if err != nil { return err } // if err := g.Wait(); err != 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 { // <-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)
// 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 { // errCh <- 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) // errCh <- 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) // errCh <- err // return gst.FlowError // } // } // } // }
// return gst.FlowOK // }, // }) // }(i) // }
// go func() { // media.HandleBusMessages(ctx, pipeline) // cancel() // }()
// if err = pipeline.SetState(gst.StatePlaying); err != nil { // return err // } // select { // case err := <-errCh: // return err // case <-ctx.Done(): // return ctx.Err() // }
return nil}