Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
8.5 kB · 240 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241package media
import ( "bytes" "context" "path/filepath" "strings" "sync" "testing" "time"
"github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" pionmedia "github.com/pion/webrtc/v4/pkg/media"
"github.com/pion/webrtc/v4" "github.com/stretchr/testify/require" "stream.place/streamplace/pkg/crypto/signers" "stream.place/streamplace/pkg/ingestframe" "stream.place/streamplace/pkg/muxl")
// whipClientOffer builds a WHIP-style SDP offer (H264 video + Opus audio tracks),// the way a real WHIP client does, returning the client PC, its tracks, and the// offer.func whipClientOffer(t *testing.T) (*webrtc.PeerConnection, *webrtc.TrackLocalStaticSample, *webrtc.TrackLocalStaticSample, webrtc.SessionDescription) { t.Helper() pc, err := webrtc.NewPeerConnection(webrtc.Configuration{}) require.NoError(t, err) videoTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, "video", "pion") require.NoError(t, err) if _, err = pc.AddTrack(videoTrack); err != nil { t.Fatal(err) } audioTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus}, "audio", "pion") require.NoError(t, err) if _, err = pc.AddTrack(audioTrack); err != nil { t.Fatal(err) } offer, err := pc.CreateOffer(nil) require.NoError(t, err) require.NoError(t, pc.SetLocalDescription(offer)) return pc, videoTrack, audioTrack, offer}
// TestWHIPWorkerAnswersOffer verifies the WHIP worker's answer back-channel: from// an offer it builds the PeerConnection, generates the SDP answer, and emits it// as the FIRST frame on its socket — the synchronous reply main returns to the// WHIP client. (Media flow → signed segments rides the same webRTCIngestPipeline// the in-process path uses, plus the transcoder/frame machinery the fMP4 ingest tests// already cover.)func TestWHIPWorkerAnswersOffer(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel()
ms := newBareSegmentSigner(t) keyPEM, err := signers.MarshalES256KPrivateKeyPEM(ms.Signer) require.NoError(t, err) manifest, err := ms.buildManifest(ctx, time.Now().UnixMilli()) require.NoError(t, err)
clientPC, _, _, offer := whipClientOffer(t) defer clientPC.Close()
sock := filepath.Join(t.TempDir(), "whip.sock") cfg := IngestWorkerConfig{ StreamerDID: ms.Streamer(), KeyPEM: keyPEM, CertPEM: ms.Cert, Manifest: manifest, NodeCertPEM: ms.Cert, NodeKeyPEM: keyPEM, BroadcasterHost: "test.example.com", SocketPath: sock, Transport: IngestTransportWHIP, OfferSDP: offer.SDP, }
serveDone := make(chan error, 1) go func() { serveDone <- ServeWHIPIngestWorkerSocket(ctx, cfg) }()
// Connect to the worker socket (retry until up) and read the Answer frame. dctx, dcancel := context.WithTimeout(ctx, 20*time.Second) defer dcancel() conn, derr := dialWorkerSocket(dctx, sock) require.NoError(t, derr) defer conn.Close()
answerSDP, rerr := readWHIPAnswer(ingestframe.NewReader(conn)) require.NoError(t, rerr, "worker emits an SDP answer as its first frame") require.Contains(t, answerSDP, "v=0", "valid SDP answer")
// The answer must apply cleanly as the client's remote description — i.e. it's // a real, negotiated answer to the offer. require.NoError(t, clientPC.SetRemoteDescription(webrtc.SessionDescription{ Type: webrtc.SDPTypeAnswer, SDP: answerSDP, }), "answer applies as the client's remote description")
t.Logf("whip worker produced a %d-byte SDP answer", len(answerSDP))
cancel() select { case <-serveDone: case <-time.After(25 * time.Second): t.Fatal("worker did not exit after cancel") }}
// produceWHIPMedia streams synthetic H264 + Opus into the WHIP client's tracks// via a gst encode pipeline until ctx is cancelled — i.e. a real WHIP publisher.func produceWHIPMedia(t *testing.T, ctx context.Context, video, audio *webrtc.TrackLocalStaticSample) { t.Helper() desc := strings.Join([]string{ "videotestsrc is-live=true ! video/x-raw,width=320,height=240,framerate=30/1 ! x264enc key-int-max=15 tune=zerolatency speed-preset=ultrafast ! h264parse ! video/x-h264,stream-format=byte-stream,alignment=au ! appsink name=vsink", "audiotestsrc is-live=true ! audioconvert ! audioresample ! opusenc ! opusparse ! appsink name=asink", }, "\n") pipeline, err := gst.NewPipelineFromString(desc) require.NoError(t, err)
pump := func(name string, track *webrtc.TrackLocalStaticSample, dur time.Duration) { ele, gerr := pipeline.GetElementByName(name) require.NoError(t, gerr) app.SinkFromElement(ele).SetCallbacks(&app.SinkCallbacks{ NewSampleFunc: func(sink *app.Sink) gst.FlowReturn { sample := sink.PullSample() if sample == nil { return gst.FlowEOS } buf := sample.GetBuffer() data := buf.Map(gst.MapRead).Bytes() buf.Unmap() if werr := track.WriteSample(pionmedia.Sample{Data: data, Duration: dur}); werr != nil { return gst.FlowError } return gst.FlowOK }, }) } pump("vsink", video, time.Second/30) pump("asink", audio, 20*time.Millisecond)
go func() { <-ctx.Done() _ = pipeline.SetState(gst.StateNull) }() require.NoError(t, pipeline.SetState(gst.StatePlaying))}
// TestWHIPWorkerLoopback is the full WHIP media path: a pion client offers,// connects to the worker (which owns the PeerConnection), and streams real// H264+Opus RTP; the worker must mux+sign+transcode it and serve a valid signed// dual-codec segment over its socket — the WHIP parity of the fMP4 ingest worker e2e// test.func TestWHIPWorkerLoopback(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel()
ms := newBareSegmentSigner(t) keyPEM, err := signers.MarshalES256KPrivateKeyPEM(ms.Signer) require.NoError(t, err) manifest, err := ms.buildManifest(ctx, time.Now().UnixMilli()) require.NoError(t, err)
clientPC, videoTrack, audioTrack, offer := whipClientOffer(t) defer clientPC.Close()
connected := make(chan struct{}) var once sync.Once clientPC.OnConnectionStateChange(func(s webrtc.PeerConnectionState) { if s == webrtc.PeerConnectionStateConnected { once.Do(func() { close(connected) }) } })
sock := filepath.Join(t.TempDir(), "whip.sock") cfg := IngestWorkerConfig{ StreamerDID: ms.Streamer(), KeyPEM: keyPEM, CertPEM: ms.Cert, Manifest: manifest, NodeCertPEM: ms.Cert, NodeKeyPEM: keyPEM, BroadcasterHost: "test.example.com", SocketPath: sock, Transport: IngestTransportWHIP, OfferSDP: offer.SDP, } serveDone := make(chan error, 1) go func() { serveDone <- ServeWHIPIngestWorkerSocket(ctx, cfg) }()
dctx, dcancel := context.WithTimeout(ctx, 20*time.Second) defer dcancel() conn, derr := dialWorkerSocket(dctx, sock) require.NoError(t, derr) defer conn.Close()
// One Reader for the whole connection: the streaming decoder reads ahead, so // the Answer and the segments after it must come through the same Reader (the // production WHIP path does the same). fr := ingestframe.NewReader(conn) answerSDP, rerr := readWHIPAnswer(fr) require.NoError(t, rerr) require.NoError(t, clientPC.SetRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeAnswer, SDP: answerSDP}))
select { case <-connected: case <-time.After(20 * time.Second): t.Fatal("client PC did not connect to the worker") }
produceWHIPMedia(t, ctx, videoTrack, audioTrack)
// Read signed segments; require at least one valid dual-codec one. _ = conn.SetReadDeadline(time.Now().Add(45 * time.Second)) var segs int for segs == 0 { typ, payload, ferr := fr.ReadFrame() require.NoError(t, ferr, "reading worker frames") if typ != ingestframe.Segment { continue } out, verr := muxl.RunMuxlVerify(ctx, bytes.NewReader(payload)) require.NoError(t, verr) require.NotContains(t, out, `"validation_state":"Invalid"`, "segment must validate") segs++ } _ = conn.SetReadDeadline(time.Time{}) t.Logf("whip worker produced %d signed segment(s) from real RTP media", segs)
// Close the connection before tearing down: in production main's connection // breaks on shutdown, which detaches the worker's frame server so it drains // into its buffer instead of blocking on an unread socket. conn.Close() cancel() select { case <-serveDone: case <-time.After(25 * time.Second): t.Fatal("worker did not exit after cancel") }}