OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

webrtc: rename incoming/outgoing into inbound/outbound (#5816)

Browse Source
This commit is contained in:
Alessandro Ros
2026-06-02 09:15:49 +02:00
committed by GitHub
parent 0f5a76e253
commit 7ab03cdc58
22 changed files with 361 additions and 363 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/api_test.go
+2 -2
View File
@@ -1231,7 +1231,7 @@ func TestAPIProtocolKick(t *testing.T) {
u, err := url.Parse("http://localhost:8889/mypath/whip")
require.NoError(t, err)
track := &webrtc.OutgoingTrack{
track := &webrtc.OutboundTrack{
Caps: pwebrtc.RTPCodecCapability{
MimeType: pwebrtc.MimeTypeH264,
ClockRate: 90000,
@@ -1244,7 +1244,7 @@ func TestAPIProtocolKick(t *testing.T) {
URL: u,
Log: test.NilLogger,
Publish: true,
OutgoingTracks: []*webrtc.OutgoingTrack{track},
OutboundTracks: []*webrtc.OutboundTrack{track},
}
err = c.Initialize(context.Background())
internal/core/metrics_test.go
+2 -2
View File
@@ -370,7 +370,7 @@ webrtc_sessions_rtcp_packets_sent 0
defer tr2.CloseIdleConnections()
hc2 := &http.Client{Transport: tr2}
track := &webrtc.OutgoingTrack{
track := &webrtc.OutboundTrack{
Caps: pwebrtc.RTPCodecCapability{
MimeType: pwebrtc.MimeTypeH264,
ClockRate: 90000,
@@ -383,7 +383,7 @@ webrtc_sessions_rtcp_packets_sent 0
URL: su,
Log: test.NilLogger,
Publish: true,
OutgoingTracks: []*webrtc.OutgoingTrack{track},
OutboundTracks: []*webrtc.OutboundTrack{track},
}
err2 = s.Initialize(context.Background())
internal/protocols/webrtc/from_stream.go
+17 -17
View File
@@ -64,12 +64,12 @@ func timestampToDuration(t int64, clockRate int) time.Duration {
func setupVideoTrack(
desc *description.Session,
r *stream.Reader,
) (*OutgoingTrack, error) {
) (*OutboundTrack, error) {
var av1Format *format.AV1
media := desc.FindFormat(&av1Format)
if av1Format != nil { //nolint:dupl
track := &OutgoingTrack{
track := &OutboundTrack{
Caps: webrtc.RTPCodecCapability{
MimeType: webrtc.MimeTypeAV1,
ClockRate: 90000,
@@ -114,7 +114,7 @@ func setupVideoTrack(
media = desc.FindFormat(&vp9Format)
if vp9Format != nil {
track := &OutgoingTrack{
track := &OutboundTrack{
Caps: webrtc.RTPCodecCapability{
MimeType: webrtc.MimeTypeVP9,
ClockRate: 90000,
@@ -161,7 +161,7 @@ func setupVideoTrack(
media = desc.FindFormat(&vp8Format)
if vp8Format != nil { //nolint:dupl
track := &OutgoingTrack{
track := &OutboundTrack{
Caps: webrtc.RTPCodecCapability{
MimeType: webrtc.MimeTypeVP8,
ClockRate: 90000,
@@ -206,7 +206,7 @@ func setupVideoTrack(
media = desc.FindFormat(&h265Format)
if h265Format != nil { //nolint:dupl
track := &OutgoingTrack{
track := &OutboundTrack{
Caps: webrtc.RTPCodecCapability{
MimeType: webrtc.MimeTypeH265,
ClockRate: 90000,
@@ -262,7 +262,7 @@ func setupVideoTrack(
media = desc.FindFormat(&h264Format)
if h264Format != nil { //nolint:dupl
track := &OutgoingTrack{
track := &OutboundTrack{
Caps: webrtc.RTPCodecCapability{
MimeType: webrtc.MimeTypeH264,
ClockRate: 90000,
@@ -320,7 +320,7 @@ func setupVideoTrack(
func setupAudioTrack(
desc *description.Session,
r *stream.Reader,
) (*OutgoingTrack, error) {
) (*OutboundTrack, error) {
var opusFormat *format.Opus
media := desc.FindFormat(&opusFormat)
@@ -354,7 +354,7 @@ func setupAudioTrack(
return nil, fmt.Errorf("unsupported channel count: %d", opusFormat.ChannelCount)
}
track := &OutgoingTrack{
track := &OutboundTrack{
Caps: caps,
}
@@ -395,7 +395,7 @@ func setupAudioTrack(
media = desc.FindFormat(&g722Format)
if g722Format != nil {
track := &OutgoingTrack{
track := &OutboundTrack{
Caps: webrtc.RTPCodecCapability{
MimeType: webrtc.MimeTypeG722,
ClockRate: 8000,
@@ -470,7 +470,7 @@ func setupAudioTrack(
}
}
track := &OutgoingTrack{
track := &OutboundTrack{
Caps: caps,
}
@@ -583,7 +583,7 @@ func setupAudioTrack(
return nil, fmt.Errorf("unsupported channel count: %d", lpcmFormat.ChannelCount)
}
track := &OutgoingTrack{
track := &OutboundTrack{
Caps: webrtc.RTPCodecCapability{
MimeType: mimeTypeL16,
ClockRate: uint32(lpcmFormat.ClockRate()),
@@ -644,12 +644,12 @@ func setupAudioTrack(
func setupKLVDataChannel(
desc *description.Session,
r *stream.Reader,
) (*OutgoingDataChannel, error) {
) (*OutboundDataChannel, error) {
var klvFormat *format.KLV
media := desc.FindFormat(&klvFormat)
if klvFormat != nil {
dataChan := &OutgoingDataChannel{
dataChan := &OutboundDataChannel{
Label: "KLV",
}
@@ -683,7 +683,7 @@ func FromStream(
}
if videoTrack != nil {
pc.OutgoingTracks = append(pc.OutgoingTracks, videoTrack)
pc.OutboundTracks = append(pc.OutboundTracks, videoTrack)
}
audioTrack, err := setupAudioTrack(desc, r)
@@ -692,7 +692,7 @@ func FromStream(
}
if audioTrack != nil {
pc.OutgoingTracks = append(pc.OutgoingTracks, audioTrack)
pc.OutboundTracks = append(pc.OutboundTracks, audioTrack)
}
klvDataChan, err := setupKLVDataChannel(desc, r)
@@ -701,10 +701,10 @@ func FromStream(
}
if klvDataChan != nil {
pc.OutgoingDataChannels = append(pc.OutgoingDataChannels, klvDataChan)
pc.OutboundDataChannels = append(pc.OutboundDataChannels, klvDataChan)
}
if len(pc.OutgoingTracks) == 0 && len(pc.OutgoingDataChannels) == 0 {
if len(pc.OutboundTracks) == 0 && len(pc.OutboundDataChannels) == 0 {
return errNoSupportedCodecsFrom
}
internal/protocols/webrtc/from_stream_test.go
+5 -5
View File
@@ -80,7 +80,7 @@ func TestFromStream(t *testing.T) {
err := FromStream(desc, r, pc)
require.NoError(t, err)
require.Equal(t, ca.webrtcCaps, pc.OutgoingTracks[0].Caps)
require.Equal(t, ca.webrtcCaps, pc.OutboundTracks[0].Caps)
})
}
}
@@ -186,10 +186,10 @@ func TestFromStreamResampleOpus(t *testing.T) {
}},
})
err = pc1.GatherIncomingTracks(2 * time.Second)
err = pc1.GatherInboundTracks(2 * time.Second)
require.NoError(t, err)
tracks := pc1.IncomingTracks()
tracks := pc1.InboundTracks()
done := make(chan struct{})
n := 0
@@ -302,10 +302,10 @@ func TestFromStreamResampleOpusAbsoluteTimestamp(t *testing.T) {
}},
})
err = pcReader.GatherIncomingTracks(2 * time.Second)
err = pcReader.GatherInboundTracks(2 * time.Second)
require.NoError(t, err)
tracks := pcReader.IncomingTracks()
tracks := pcReader.InboundTracks()
require.Len(t, tracks, 1)
done := make(chan struct{})
internal/protocols/webrtc/incoming_track.go → internal/protocols/webrtc/inbound_track.go
+12 -12
View File
@@ -18,7 +18,7 @@ const (
mimeTypeL16 = "audio/L16"
)
func incomingTrackTWCCExtensionID(params webrtc.RTPParameters) uint8 {
func inboundTrackTWCCExtensionID(params webrtc.RTPParameters) uint8 {
for _, ext := range params.HeaderExtensions {
if ext.URI == twccExtensionURI {
return uint8(ext.ID)
@@ -243,8 +243,8 @@ var incomingAudioCodecs = []webrtc.RTPCodecParameters{
},
}
// IncomingTrack is an incoming track.
type IncomingTrack struct {
// InboundTrack is an incoming track.
type InboundTrack struct {
OnPacketRTP func(*rtp.Packet)
track *webrtc.TrackRemote
@@ -258,12 +258,12 @@ type IncomingTrack struct {
rtpReceiver *rtpreceiver.Receiver
}
func (t *IncomingTrack) initialize() {
func (t *InboundTrack) initialize() {
t.OnPacketRTP = func(*rtp.Packet) {}
t.twccExtID = incomingTrackTWCCExtensionID(t.receiver.GetParameters())
t.twccExtID = inboundTrackTWCCExtensionID(t.receiver.GetParameters())
}
func (t *IncomingTrack) stripTWCCExtension(pkt *rtp.Packet) {
func (t *InboundTrack) stripTWCCExtension(pkt *rtp.Packet) {
if t.twccExtID == 0 || pkt.GetExtension(t.twccExtID) == nil {
return
}
@@ -280,21 +280,21 @@ func (t *IncomingTrack) stripTWCCExtension(pkt *rtp.Packet) {
}
// Codec returns the track codec.
func (t *IncomingTrack) Codec() webrtc.RTPCodecParameters {
func (t *InboundTrack) Codec() webrtc.RTPCodecParameters {
return t.track.Codec()
}
// ClockRate returns the clock rate. Needed by rtptime.GlobalDecoder
func (t *IncomingTrack) ClockRate() int {
func (t *InboundTrack) ClockRate() int {
return int(t.track.Codec().ClockRate)
}
// PTSEqualsDTS returns whether PTS equals DTS. Needed by rtptime.GlobalDecoder
func (*IncomingTrack) PTSEqualsDTS(*rtp.Packet) bool {
func (*InboundTrack) PTSEqualsDTS(*rtp.Packet) bool {
return true
}
func (t *IncomingTrack) start() {
func (t *InboundTrack) start() {
t.inboundRTPPacketsLost = &counterdumper.Dumper{
OnReport: func(val uint64) {
t.log.Log(logger.Warn, "%d RTP %s lost",
@@ -393,11 +393,11 @@ func (t *IncomingTrack) start() {
}
// PacketNTP returns the packet NTP.
func (t *IncomingTrack) PacketNTP(pkt *rtp.Packet) (time.Time, bool) {
func (t *InboundTrack) PacketNTP(pkt *rtp.Packet) (time.Time, bool) {
return t.rtpReceiver.PacketNTP(pkt.Timestamp)
}
func (t *IncomingTrack) close() {
func (t *InboundTrack) close() {
if t.inboundRTPPacketsLost != nil {
t.inboundRTPPacketsLost.Stop()
}
internal/protocols/webrtc/outgoing_data_channel.go → internal/protocols/webrtc/outbound_data_channel.go
+4 -4
View File
@@ -4,14 +4,14 @@ import (
"github.com/pion/webrtc/v4"
)
// OutgoingDataChannel is an outgoing data channel.
type OutgoingDataChannel struct {
// OutboundDataChannel is an outgoing data channel.
type OutboundDataChannel struct {
Label string
dataChan *webrtc.DataChannel
}
func (c *OutgoingDataChannel) setup(p *PeerConnection) error {
func (c *OutboundDataChannel) setup(p *PeerConnection) error {
var err error
c.dataChan, err = p.wr.CreateDataChannel(c.Label, &webrtc.DataChannelInit{
Ordered: new(false),
@@ -24,6 +24,6 @@ func (c *OutgoingDataChannel) setup(p *PeerConnection) error {
}
// Write writes data to the channel.
func (c *OutgoingDataChannel) Write(data []byte) {
func (c *OutboundDataChannel) Write(data []byte) {
c.dataChan.Send(data) //nolint:errcheck
}
internal/protocols/webrtc/outgoing_track.go → internal/protocols/webrtc/outbound_track.go
+7 -7
View File
@@ -10,8 +10,8 @@ import (
"github.com/pion/webrtc/v4"
)
// OutgoingTrack is an outgoing track.
type OutgoingTrack struct {
// OutboundTrack is an outgoing track.
type OutboundTrack struct {
Caps webrtc.RTPCodecCapability
track *webrtc.TrackLocalStaticRTP
@@ -19,11 +19,11 @@ type OutgoingTrack struct {
rtcpSender *rtpsender.Sender
}
func (t *OutgoingTrack) isVideo() bool {
func (t *OutboundTrack) isVideo() bool {
return strings.Split(t.Caps.MimeType, "/")[0] == "video"
}
func (t *OutgoingTrack) setup(p *PeerConnection) error {
func (t *OutboundTrack) setup(p *PeerConnection) error {
var trackID string
if t.isVideo() {
trackID = "video"
@@ -77,19 +77,19 @@ func (t *OutgoingTrack) setup(p *PeerConnection) error {
return nil
}
func (t *OutgoingTrack) close() {
func (t *OutboundTrack) close() {
if t.rtcpSender != nil {
t.rtcpSender.Close()
}
}
// WriteRTP writes a RTP packet.
func (t *OutgoingTrack) WriteRTP(pkt *rtp.Packet) error {
func (t *OutboundTrack) WriteRTP(pkt *rtp.Packet) error {
return t.WriteRTPWithNTP(pkt, time.Now())
}
// WriteRTPWithNTP writes a RTP packet.
func (t *OutgoingTrack) WriteRTPWithNTP(pkt *rtp.Packet, ntp time.Time) error {
func (t *OutboundTrack) WriteRTPWithNTP(pkt *rtp.Packet, ntp time.Time) error {
// use right SSRC in packet to make rtcpSender work
pkt.SSRC = t.ssrc
internal/protocols/webrtc/peer_connection.go
+26 -26
View File
@@ -161,19 +161,19 @@ type PeerConnection struct {
AdditionalHosts []string
STUNGatherTimeout time.Duration
Publish bool
OutgoingTracks []*OutgoingTrack
OutgoingDataChannels []*OutgoingDataChannel
OutboundTracks []*OutboundTrack
OutboundDataChannels []*OutboundDataChannel
Log logger.Writer
wr *webrtc.PeerConnection
ctx context.Context
ctxCancel context.CancelFunc
readingStarted atomic.Int64
incomingTracks []*IncomingTrack
inboundTracks []*InboundTrack
statsInterceptor *statsInterceptor
newLocalCandidate chan *webrtc.ICECandidateInit
incomingTrack chan trackRecvPair
inboundTrack chan trackRecvPair
stateMutex sync.Mutex
state webrtc.PeerConnectionState
@@ -232,7 +232,7 @@ func (co *PeerConnection) Start() error {
if co.Publish {
videoSetupped := false
audioSetupped := false
for _, tr := range co.OutgoingTracks {
for _, tr := range co.OutboundTracks {
if tr.isVideo() {
videoSetupped = true
} else {
@@ -243,7 +243,7 @@ func (co *PeerConnection) Start() error {
// When audio is not used, a track has to be present anyway,
// otherwise video is not displayed on Firefox and Chrome.
if !audioSetupped {
co.OutgoingTracks = append(co.OutgoingTracks, &OutgoingTrack{
co.OutboundTracks = append(co.OutboundTracks, &OutboundTrack{
Caps: webrtc.RTPCodecCapability{
MimeType: webrtc.MimeTypePCMU,
ClockRate: 8000,
@@ -251,7 +251,7 @@ func (co *PeerConnection) Start() error {
})
}
for i, tr := range co.OutgoingTracks {
for i, tr := range co.OutboundTracks {
var codecType webrtc.RTPCodecType
if tr.isVideo() {
codecType = webrtc.RTPCodecTypeVideo
@@ -328,12 +328,12 @@ func (co *PeerConnection) Start() error {
co.newLocalCandidate = make(chan *webrtc.ICECandidateInit)
co.stateChanged = make(chan struct{})
co.gatheringDone = make(chan struct{})
co.incomingTrack = make(chan trackRecvPair)
co.inboundTrack = make(chan trackRecvPair)
co.done = make(chan struct{})
co.chStartReading = make(chan struct{})
if co.Publish {
for _, tr := range co.OutgoingTracks {
for _, tr := range co.OutboundTracks {
err = tr.setup(co)
if err != nil {
co.wr.GracefulClose() //nolint:errcheck
@@ -341,7 +341,7 @@ func (co *PeerConnection) Start() error {
}
}
for _, dc := range co.OutgoingDataChannels {
for _, dc := range co.OutboundDataChannels {
err = dc.setup(co)
if err != nil {
co.wr.GracefulClose() //nolint:errcheck
@@ -367,7 +367,7 @@ func (co *PeerConnection) Start() error {
co.wr.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
select {
case co.incomingTrack <- trackRecvPair{track, receiver}:
case co.inboundTrack <- trackRecvPair{track, receiver}:
case <-co.ctx.Done():
}
})
@@ -429,10 +429,10 @@ func (co *PeerConnection) run() {
defer close(co.done)
defer func() {
for _, track := range co.incomingTracks {
for _, track := range co.inboundTracks {
track.close()
}
for _, track := range co.OutgoingTracks {
for _, track := range co.OutboundTracks {
track.close()
}
@@ -448,7 +448,7 @@ func (co *PeerConnection) run() {
for {
select {
case <-co.chStartReading:
for _, track := range co.incomingTracks {
for _, track := range co.inboundTracks {
track.start()
}
co.readingStarted.Store(1)
@@ -762,8 +762,8 @@ outer:
return nil
}
// GatherIncomingTracks gathers incoming tracks.
func (co *PeerConnection) GatherIncomingTracks(timeout time.Duration) error {
// GatherInboundTracks gathers incoming tracks.
func (co *PeerConnection) GatherInboundTracks(timeout time.Duration) error {
var sdp sdp.SessionDescription
sdp.Unmarshal([]byte(co.wr.RemoteDescription().SDP)) //nolint:errcheck
@@ -775,13 +775,13 @@ func (co *PeerConnection) GatherIncomingTracks(timeout time.Duration) error {
for {
select {
case <-t.C:
if len(co.incomingTracks) != 0 {
if len(co.inboundTracks) != 0 {
return nil
}
return fmt.Errorf("deadline exceeded while waiting tracks")
case pair := <-co.incomingTrack:
t := &IncomingTrack{
case pair := <-co.inboundTrack:
t := &InboundTrack{
track: pair.track,
receiver: pair.receiver,
rid: pair.track.RID(),
@@ -789,9 +789,9 @@ func (co *PeerConnection) GatherIncomingTracks(timeout time.Duration) error {
log: co.Log,
}
t.initialize()
co.incomingTracks = append(co.incomingTracks, t)
co.inboundTracks = append(co.inboundTracks, t)
if len(co.incomingTracks) >= maxTrackCount {
if len(co.inboundTracks) >= maxTrackCount {
return nil
}
@@ -895,9 +895,9 @@ func (co *PeerConnection) GatheringDone() <-chan struct{} {
return co.gatheringDone
}
// IncomingTracks returns incoming tracks.
func (co *PeerConnection) IncomingTracks() []*IncomingTrack {
return co.incomingTracks
// InboundTracks returns incoming tracks.
func (co *PeerConnection) InboundTracks() []*InboundTrack {
return co.inboundTracks
}
// StartReading starts reading incoming tracks.
@@ -960,7 +960,7 @@ func (co *PeerConnection) Stats() *Stats {
packetsLost := uint64(0)
if co.readingStarted.Load() == 1 {
for _, tr := range co.incomingTracks {
for _, tr := range co.inboundTracks {
if recvStats := tr.rtpReceiver.Stats(); recvStats != nil {
v += recvStats.Jitter
n++
@@ -970,7 +970,7 @@ func (co *PeerConnection) Stats() *Stats {
}
}
for _, tr := range co.OutgoingTracks {
for _, tr := range co.OutboundTracks {
if sentStats := tr.rtcpSender.Stats(); sentStats != nil {
packetsSent += sentStats.Sent
}
internal/protocols/webrtc/peer_connection_test.go
+15 -15
View File
@@ -26,7 +26,7 @@ func (nilWriter) Write(p []byte) (int, error) {
var webrtcNilLogger = logging.NewDefaultLeveledLoggerForScope("", 0, &nilWriter{})
func gatherCodecs(tracks []*IncomingTrack) []webrtc.RTPCodecParameters {
func gatherCodecs(tracks []*InboundTrack) []webrtc.RTPCodecParameters {
codecs := make([]webrtc.RTPCodecParameters, len(tracks))
for i, track := range tracks {
codecs[i] = track.Codec()
@@ -235,7 +235,7 @@ func TestPeerConnectionConnectivity(t *testing.T) {
ICETCPMux: tcpMux,
ICEServers: iceServers,
Publish: true,
OutgoingTracks: []*OutgoingTrack{{
OutboundTracks: []*OutboundTrack{{
Caps: webrtc.RTPCodecCapability{
MimeType: webrtc.MimeTypeAV1,
ClockRate: 90000,
@@ -370,10 +370,10 @@ func TestPeerConnectionRead(t *testing.T) {
require.NoError(t, err2)
}()
err = reader.GatherIncomingTracks(2 * time.Second)
err = reader.GatherInboundTracks(2 * time.Second)
require.NoError(t, err)
codecs := gatherCodecs(reader.IncomingTracks())
codecs := gatherCodecs(reader.InboundTracks())
sort.Slice(codecs, func(i, j int) bool {
return codecs[i].PayloadType < codecs[j].PayloadType
@@ -556,10 +556,10 @@ func TestPeerConnectionReadSimulcast(t *testing.T) {
}
}()
err = reader.GatherIncomingTracks(5 * time.Second)
err = reader.GatherInboundTracks(5 * time.Second)
require.NoError(t, err)
tracks := reader.IncomingTracks()
tracks := reader.InboundTracks()
codecs := gatherCodecs(tracks)
require.Equal(t, 3, len(codecs))
@@ -651,10 +651,10 @@ func TestPeerConnectionStripIncomingTWCC(t *testing.T) {
}
}()
err = reader.GatherIncomingTracks(5 * time.Second)
err = reader.GatherInboundTracks(5 * time.Second)
require.NoError(t, err)
tracks := reader.IncomingTracks()
tracks := reader.InboundTracks()
require.Len(t, tracks, 1)
done := make(chan struct{})
@@ -685,7 +685,7 @@ func TestPeerConnectionPublishRead(t *testing.T) {
LocalRandomUDP: true,
IPsFromInterfaces: true,
Publish: true,
OutgoingTracks: []*OutgoingTrack{
OutboundTracks: []*OutboundTrack{
{
Caps: webrtc.RTPCodecCapability{
MimeType: webrtc.MimeTypeH264,
@@ -721,7 +721,7 @@ func TestPeerConnectionPublishRead(t *testing.T) {
err = pc2.WaitUntilConnected(10 * time.Second)
require.NoError(t, err)
for _, track := range pc2.OutgoingTracks {
for _, track := range pc2.OutboundTracks {
err = track.WriteRTP(&rtp.Packet{
Header: rtp.Header{
Version: 2,
@@ -736,10 +736,10 @@ func TestPeerConnectionPublishRead(t *testing.T) {
require.NoError(t, err)
}
err = pc1.GatherIncomingTracks(2 * time.Second)
err = pc1.GatherInboundTracks(2 * time.Second)
require.NoError(t, err)
codecs := gatherCodecs(pc1.IncomingTracks())
codecs := gatherCodecs(pc1.InboundTracks())
sort.Slice(codecs, func(i, j int) bool {
return codecs[i].PayloadType < codecs[j].PayloadType
@@ -784,7 +784,7 @@ func TestPeerConnectionFallbackCodecs(t *testing.T) {
LocalRandomUDP: true,
IPsFromInterfaces: true,
Publish: true,
OutgoingTracks: []*OutgoingTrack{{
OutboundTracks: []*OutboundTrack{{
Caps: webrtc.RTPCodecCapability{
MimeType: webrtc.MimeTypeAV1,
ClockRate: 90000,
@@ -860,7 +860,7 @@ func TestPeerConnectionPublishDataChannel(t *testing.T) {
LocalRandomUDP: true,
IPsFromInterfaces: true,
Publish: true,
OutgoingDataChannels: []*OutgoingDataChannel{
OutboundDataChannels: []*OutboundDataChannel{
{
Label: "test-channel",
},
@@ -882,7 +882,7 @@ func TestPeerConnectionPublishDataChannel(t *testing.T) {
<-dataChanCreated
pc2.OutgoingDataChannels[0].Write([]byte("test data"))
pc2.OutboundDataChannels[0].Write([]byte("test data"))
<-dataReceived
}
internal/protocols/webrtc/to_stream.go
+1 -1
View File
@@ -40,7 +40,7 @@ func ToStream(
timeDecoder := &rtptime.GlobalDecoder{}
timeDecoder.Initialize()
for _, track := range pc.incomingTracks {
for _, track := range pc.inboundTracks {
var typ description.MediaType
var forma format.Format
internal/protocols/webrtc/to_stream_test.go
+3 -3
View File
@@ -338,7 +338,7 @@ func TestToStream(t *testing.T) {
LocalRandomUDP: true,
IPsFromInterfaces: true,
Publish: true,
OutgoingTracks: []*OutgoingTrack{{
OutboundTracks: []*OutboundTrack{{
Caps: ca.webrtcCaps,
}},
Log: test.NilLogger,
@@ -385,7 +385,7 @@ func TestToStream(t *testing.T) {
err = pc2.WaitUntilConnected(10 * time.Second)
require.NoError(t, err)
err = pc1.OutgoingTracks[0].WriteRTP(&rtp.Packet{
err = pc1.OutboundTracks[0].WriteRTP(&rtp.Packet{
Header: rtp.Header{
Version: 2,
Marker: true,
@@ -398,7 +398,7 @@ func TestToStream(t *testing.T) {
})
require.NoError(t, err)
err = pc2.GatherIncomingTracks(2 * time.Second)
err = pc2.GatherInboundTracks(2 * time.Second)
require.NoError(t, err)
var subStream *stream.SubStream
internal/protocols/whip/client.go
+6 -6
View File
@@ -77,7 +77,7 @@ func offerAndCandidateToSDPFragment(
type Client struct {
URL *url.URL
Publish bool
OutgoingTracks []*webrtc.OutgoingTrack
OutboundTracks []*webrtc.OutboundTrack
HTTPClient *http.Client
BearerToken string
UDPReadBufferSize uint
@@ -114,7 +114,7 @@ func (c *Client) Initialize(ctx context.Context) error {
IPsFromInterfaces: true,
Publish: c.Publish,
STUNGatherTimeout: c.STUNGatherTimeout,
OutgoingTracks: c.OutgoingTracks,
OutboundTracks: c.OutboundTracks,
Log: c.Log,
}
err = c.pc.Start()
@@ -199,7 +199,7 @@ func (c *Client) initializeInner(ctx context.Context) error {
}
if !c.Publish {
err = c.pc.GatherIncomingTracks(c.TrackGatherTimeout)
err = c.pc.GatherInboundTracks(c.TrackGatherTimeout)
if err != nil {
c.deleteSession(context.Background()) //nolint:errcheck
return err
@@ -247,9 +247,9 @@ func (c *Client) PeerConnection() *webrtc.PeerConnection {
return c.pc
}
// IncomingTracks returns incoming tracks.
func (c *Client) IncomingTracks() []*webrtc.IncomingTrack {
return c.pc.IncomingTracks()
// InboundTracks returns incoming tracks.
func (c *Client) InboundTracks() []*webrtc.InboundTrack {
return c.pc.InboundTracks()
}
// StartReading starts reading all incoming tracks.
internal/protocols/whip/client_test.go
+27 -27
View File
@@ -24,7 +24,7 @@ func whipOffer(body []byte) *pwebrtc.SessionDescription {
}
}
func gatherCodecs(tracks []*webrtc.IncomingTrack) []pwebrtc.RTPCodecParameters {
func gatherCodecs(tracks []*webrtc.InboundTrack) []pwebrtc.RTPCodecParameters {
codecs := make([]pwebrtc.RTPCodecParameters, len(tracks))
for i, track := range tracks {
codecs[i] = track.Codec()
@@ -38,11 +38,11 @@ func TestClientRead(t *testing.T) {
"video+audio",
} {
t.Run(ca, func(t *testing.T) {
var outgoingTracks []*webrtc.OutgoingTrack
var outboundTracks []*webrtc.OutboundTrack
switch ca {
case "audio":
outgoingTracks = []*webrtc.OutgoingTrack{{
outboundTracks = []*webrtc.OutboundTrack{{
Caps: pwebrtc.RTPCodecCapability{
MimeType: "audio/opus",
ClockRate: 48000,
@@ -51,7 +51,7 @@ func TestClientRead(t *testing.T) {
}}
case "video+audio":
outgoingTracks = []*webrtc.OutgoingTrack{
outboundTracks = []*webrtc.OutboundTrack{
{
Caps: pwebrtc.RTPCodecCapability{
MimeType: "video/H264",
@@ -72,7 +72,7 @@ func TestClientRead(t *testing.T) {
LocalRandomUDP: true,
IPsFromInterfaces: true,
Publish: true,
OutgoingTracks: outgoingTracks,
OutboundTracks: outboundTracks,
Log: test.NilLogger,
}
err := pc.Start()
@@ -116,7 +116,7 @@ func TestClientRead(t *testing.T) {
err3 := pc.WaitUntilConnected(10 * time.Second)
require.NoError(t, err3)
for _, track := range outgoingTracks {
for _, track := range outboundTracks {
err3 = track.WriteRTP(&rtp.Packet{
Header: rtp.Header{
Version: 2,
@@ -168,7 +168,7 @@ func TestClientRead(t *testing.T) {
require.NoError(t, err)
defer cl.Close() //nolint:errcheck
codecs := gatherCodecs(cl.IncomingTracks())
codecs := gatherCodecs(cl.InboundTracks())
switch ca {
case "audio":
@@ -215,12 +215,12 @@ func TestClientRead(t *testing.T) {
}, codecs)
}
recv := make([]chan struct{}, len(outgoingTracks))
for i := range outgoingTracks {
recv := make([]chan struct{}, len(outboundTracks))
for i := range outboundTracks {
recv[i] = make(chan struct{})
}
for i, track := range cl.IncomingTracks() {
for i, track := range cl.InboundTracks() {
ci := i
track.OnPacketRTP = func(_ *rtp.Packet) {
close(recv[ci])
@@ -286,10 +286,10 @@ func TestClientPublish(t *testing.T) {
err3 := pc.WaitUntilConnected(10 * time.Second)
require.NoError(t, err3)
err3 = pc.GatherIncomingTracks(2 * time.Second)
err3 = pc.GatherInboundTracks(2 * time.Second)
require.NoError(t, err3)
codecs := gatherCodecs(pc.IncomingTracks())
codecs := gatherCodecs(pc.InboundTracks())
switch ca {
case "audio":
@@ -333,7 +333,7 @@ func TestClientPublish(t *testing.T) {
}, codecs)
}
for i, track := range pc.IncomingTracks() {
for i, track := range pc.InboundTracks() {
ci := i
track.OnPacketRTP = func(_ *rtp.Packet) {
close(recv[ci])
@@ -370,11 +370,11 @@ func TestClientPublish(t *testing.T) {
u, err := url.Parse("http://localhost:9005/my/resource")
require.NoError(t, err)
var outgoingTracks []*webrtc.OutgoingTrack
var outboundTracks []*webrtc.OutboundTrack
switch ca {
case "audio":
outgoingTracks = []*webrtc.OutgoingTrack{{
outboundTracks = []*webrtc.OutboundTrack{{
Caps: pwebrtc.RTPCodecCapability{
MimeType: "audio/opus",
ClockRate: 48000,
@@ -383,7 +383,7 @@ func TestClientPublish(t *testing.T) {
}}
case "video+audio":
outgoingTracks = []*webrtc.OutgoingTrack{
outboundTracks = []*webrtc.OutboundTrack{
{
Caps: pwebrtc.RTPCodecCapability{
MimeType: "video/H264",
@@ -400,15 +400,15 @@ func TestClientPublish(t *testing.T) {
}
}
recv = make([]chan struct{}, len(outgoingTracks))
for i := range outgoingTracks {
recv = make([]chan struct{}, len(outboundTracks))
for i := range outboundTracks {
recv[i] = make(chan struct{})
}
cl := &Client{
URL: u,
Publish: true,
OutgoingTracks: outgoingTracks,
OutboundTracks: outboundTracks,
HTTPClient: &http.Client{},
Log: test.NilLogger,
}
@@ -416,7 +416,7 @@ func TestClientPublish(t *testing.T) {
require.NoError(t, err)
defer cl.Close() //nolint:errcheck
for _, track := range cl.OutgoingTracks {
for _, track := range cl.OutboundTracks {
err = track.WriteRTP(&rtp.Packet{
Header: rtp.Header{
Version: 2,
@@ -481,7 +481,7 @@ func TestClientBearerToken(t *testing.T) {
u, err := url.Parse("http://localhost:9005/my/resource")
require.NoError(t, err)
outgoingTracks := []*webrtc.OutgoingTrack{{
outboundTracks := []*webrtc.OutboundTrack{{
Caps: pwebrtc.RTPCodecCapability{
MimeType: "audio/opus",
ClockRate: 48000,
@@ -495,7 +495,7 @@ func TestClientBearerToken(t *testing.T) {
BearerToken: "my_secret_token",
Log: test.NilLogger,
Publish: true,
OutgoingTracks: outgoingTracks,
OutboundTracks: outboundTracks,
}
err = cl.Initialize(context.Background())
require.NoError(t, err)
@@ -549,10 +549,10 @@ func TestClientNoTrickleICE(t *testing.T) {
err3 := pc.WaitUntilConnected(10 * time.Second)
require.NoError(t, err3)
err3 = pc.GatherIncomingTracks(2 * time.Second)
err3 = pc.GatherInboundTracks(2 * time.Second)
require.NoError(t, err3)
pc.IncomingTracks()[0].OnPacketRTP = func(_ *rtp.Packet) {
pc.InboundTracks()[0].OnPacketRTP = func(_ *rtp.Packet) {
close(recv)
}
@@ -583,7 +583,7 @@ func TestClientNoTrickleICE(t *testing.T) {
u, err := url.Parse("http://localhost:9005/my/resource")
require.NoError(t, err)
outgoingTracks := []*webrtc.OutgoingTrack{{
outboundTracks := []*webrtc.OutboundTrack{{
Caps: pwebrtc.RTPCodecCapability{
MimeType: "audio/opus",
ClockRate: 48000,
@@ -594,7 +594,7 @@ func TestClientNoTrickleICE(t *testing.T) {
cl := &Client{
URL: u,
Publish: true,
OutgoingTracks: outgoingTracks,
OutboundTracks: outboundTracks,
HTTPClient: &http.Client{},
Log: test.NilLogger,
}
@@ -602,7 +602,7 @@ func TestClientNoTrickleICE(t *testing.T) {
require.NoError(t, err)
defer cl.Close() //nolint:errcheck
err = outgoingTracks[0].WriteRTP(&rtp.Packet{
err = outboundTracks[0].WriteRTP(&rtp.Packet{
Header: rtp.Header{
Version: 2,
Marker: true,
internal/servers/webrtc/http_server.go
+31 -34
View File
@@ -89,9 +89,7 @@ type httpServer struct {
func (s *httpServer) initialize() error {
router := gin.New()
router.SetTrustedProxies(s.trustedProxies.ToTrustedProxies()) //nolint:errcheck
router.Use(s.middlewarePreflightRequests)
router.Use(s.onRequest)
var proto string
@@ -205,16 +203,22 @@ func (s *httpServer) onWHIPPost(ctx *gin.Context, pathName string, publish bool)
return
}
res := s.parent.newSession(webRTCNewSessionReq{
res := s.parent.newSession(newSessionReq{
pathName: pathName,
remoteAddr: httpp.RemoteAddr(ctx),
offer: offer,
publish: publish,
offer: offer,
httpRequest: ctx.Request,
})
if res.err != nil {
s.writeErrorNoLog(ctx, res.errStatusCode, res.err)
return
}
res2 := res.sx.initialRequest(initialRequestReq{})
if res2.err != nil {
var terr *auth.Error
if errors.As(res.err, &terr) {
if errors.As(res2.err, &terr) {
if terr.AskCredentials {
ctx.Header("WWW-Authenticate", `Basic realm="mediamtx"`)
s.writeErrorNoLog(ctx, http.StatusUnauthorized, fmt.Errorf("authentication error"))
@@ -230,7 +234,7 @@ func (s *httpServer) onWHIPPost(ctx *gin.Context, pathName string, publish bool)
return
}
s.writeErrorNoLog(ctx, res.errStatusCode, res.err)
s.writeErrorNoLog(ctx, res2.errStatusCode, res2.err)
return
}
@@ -252,12 +256,12 @@ func (s *httpServer) onWHIPPost(ctx *gin.Context, pathName string, publish bool)
ctx.Writer.Header()["Link"] = whip.LinkHeaderMarshal(servers)
ctx.Header("Location", sessionLocation(publish, pathName, ctx.Request.URL.RawQuery, res.sx.secret))
ctx.Writer.WriteHeader(http.StatusCreated)
ctx.Writer.Write(res.answer)
ctx.Writer.Write(res2.answer)
res.sx.Log(logger.Debug, "SDP answer:\n"+string(res.answer))
res.sx.Log(logger.Debug, "SDP answer:\n"+string(res2.answer))
}
func (s *httpServer) onWHIPPatch(ctx *gin.Context, pathName string, rawSecret string) {
func (s *httpServer) onWHIPPatch(ctx *gin.Context, rawSecret string) {
secret, err := uuid.Parse(rawSecret)
if err != nil {
s.writeErrorNoLog(ctx, http.StatusBadRequest, fmt.Errorf("invalid secret"))
@@ -282,8 +286,7 @@ func (s *httpServer) onWHIPPatch(ctx *gin.Context, pathName string, rawSecret st
return
}
res := s.parent.addSessionCandidates(webRTCAddSessionCandidatesReq{
pathName: pathName,
res := s.parent.addSessionCandidates(addSessionCandidatesReq{
secret: secret,
fragment: &frag,
})
@@ -323,16 +326,15 @@ func (s *httpServer) onWHIPPatch(ctx *gin.Context, pathName string, rawSecret st
})
}
func (s *httpServer) onWHIPDelete(ctx *gin.Context, pathName string, rawSecret string) {
func (s *httpServer) onWHIPDelete(ctx *gin.Context, rawSecret string) {
secret, err := uuid.Parse(rawSecret)
if err != nil {
s.writeErrorNoLog(ctx, http.StatusBadRequest, fmt.Errorf("invalid secret"))
return
}
err = s.parent.deleteSession(webRTCDeleteSessionReq{
pathName: pathName,
secret: secret,
err = s.parent.deleteSession(deleteSessionReq{
secret: secret,
})
if err != nil {
if errors.Is(err, ErrSessionNotFound) {
@@ -379,22 +381,6 @@ func (s *httpServer) middlewarePreflightRequests(ctx *gin.Context) {
}
func (s *httpServer) onRequest(ctx *gin.Context) {
if strings.HasSuffix(ctx.Request.URL.Path, "/publisher.js") {
ctx.Header("Cache-Control", "max-age=3600")
ctx.Header("Content-Type", "application/javascript")
ctx.Writer.WriteHeader(http.StatusOK)
ctx.Writer.Write(publisherJS)
return
}
if strings.HasSuffix(ctx.Request.URL.Path, "/reader.js") {
ctx.Header("Cache-Control", "max-age=3600")
ctx.Header("Content-Type", "application/javascript")
ctx.Writer.WriteHeader(http.StatusOK)
ctx.Writer.Write(readerJS)
return
}
// WHIP/WHEP, outside session
if m := reWHIPWHEPNoID.FindStringSubmatch(ctx.Request.URL.Path); m != nil {
switch ctx.Request.Method {
@@ -417,10 +403,10 @@ func (s *httpServer) onRequest(ctx *gin.Context) {
if m := reWHIPWHEPWithID.FindStringSubmatch(ctx.Request.URL.Path); m != nil {
switch ctx.Request.Method {
case http.MethodPatch:
s.onWHIPPatch(ctx, m[1], m[3])
s.onWHIPPatch(ctx, m[3])
case http.MethodDelete:
s.onWHIPDelete(ctx, m[1], m[3])
s.onWHIPDelete(ctx, m[3])
}
return
}
@@ -428,6 +414,18 @@ func (s *httpServer) onRequest(ctx *gin.Context) {
// static resources
if ctx.Request.Method == http.MethodGet {
switch {
case strings.HasSuffix(ctx.Request.URL.Path, "/publisher.js"):
ctx.Header("Cache-Control", "max-age=3600")
ctx.Header("Content-Type", "application/javascript")
ctx.Writer.WriteHeader(http.StatusOK)
ctx.Writer.Write(publisherJS)
case strings.HasSuffix(ctx.Request.URL.Path, "/reader.js"):
ctx.Header("Cache-Control", "max-age=3600")
ctx.Header("Content-Type", "application/javascript")
ctx.Writer.WriteHeader(http.StatusOK)
ctx.Writer.Write(readerJS)
case ctx.Request.URL.Path == "/favicon.ico":
case len(ctx.Request.URL.Path) >= 2:
@@ -443,6 +441,5 @@ func (s *httpServer) onRequest(ctx *gin.Context) {
s.onPage(ctx, ctx.Request.URL.Path[1:len(ctx.Request.URL.Path)-1], false)
}
}
return
}
}
internal/servers/webrtc/publish_index.html
+37 -39
View File
@@ -159,7 +159,6 @@
const controls = document.getElementById("controls");
const message = document.getElementById("message");
const publishButton = document.getElementById("publish-button");
let publisher = null;
const videoForm = {
device: document.getElementById("video-device"),
@@ -177,6 +176,8 @@
voice: document.getElementById("audio-voice"),
};
let publisher = null;
const setMessage = (str) => {
message.innerText = str;
};
@@ -209,6 +210,8 @@
const videoId = videoForm.device.value;
const audioId = audioForm.device.value;
let mediaPromise;
if (videoId !== "screen") {
let videoOpts = false;
@@ -236,31 +239,27 @@
}
}
navigator.mediaDevices
.getUserMedia({
video: videoOpts,
audio: audioOpts,
})
.then((stream) => onStream(stream))
.catch((err) => {
setMessage(err.toString());
});
mediaPromise = navigator.mediaDevices.getUserMedia({
video: videoOpts,
audio: audioOpts,
});
} else {
navigator.mediaDevices
.getDisplayMedia({
video: {
width: { ideal: videoForm.width.value },
height: { ideal: videoForm.height.value },
frameRate: { ideal: videoForm.framerate.value },
cursor: "always",
},
audio: true,
})
.then((stream) => onStream(stream))
.catch((err) => {
setMessage(err.toString());
});
mediaPromise = navigator.mediaDevices.getDisplayMedia({
video: {
width: { ideal: videoForm.width.value },
height: { ideal: videoForm.height.value },
frameRate: { ideal: videoForm.framerate.value },
cursor: "always",
},
audio: true,
});
}
mediaPromise
.then((stream) => onStream(stream))
.catch((err) => {
setMessage(err.toString());
});
};
const selectHasOption = (select, option) => {
@@ -350,29 +349,18 @@
});
};
const populateOptions = () => {
const loadDevices = () => {
setMessage("loading devices");
navigator.mediaDevices
return navigator.mediaDevices
.getUserMedia({ video: true, audio: true })
.then((tempStream) => {
return Promise.all([populateDevices(), populateCodecs()]).then(
() => {
// free the webcam to prevent 'NotReadableError' on Android
tempStream.getTracks().forEach((track) => track.stop());
setMessage("");
loadValuesFromQuery();
setupEventListeners();
video.style.display = "none";
controls.style.display = "flex";
},
);
})
.catch((err) => {
setMessage(err.toString());
});
};
@@ -439,8 +427,18 @@
return;
}
publishButton.addEventListener("click", onPublish);
populateOptions();
loadDevices()
.then(() => {
loadValuesFromQuery();
setupEventListeners();
video.style.display = "none";
controls.style.display = "flex";
publishButton.addEventListener("click", onPublish);
setMessage("");
})
.catch((err) => {
setMessage(err.toString());
});
});
window.addEventListener("beforeunload", () => {
internal/servers/webrtc/publisher.js
+24 -28
View File
@@ -28,35 +28,29 @@
/** WebRTC/WHIP publisher. */
class MediaMTXWebRTCPublisher {
#retryPause;
static #RETRY_PAUSE = 2000;
#conf;
#state;
#restartTimeout;
#pc;
#offerData;
#sessionUrl;
#queuedCandidates;
#state = "running";
#restartTimeout = null;
#pc = null;
#offerData = null;
#sessionUrl = null;
#queuedCandidates = [];
/**
* Create a MediaMTXWebRTCPublisher.
* @param {Conf} conf - configuration.
*/
constructor(conf) {
this.#retryPause = 2000;
this.#conf = conf;
this.#state = "running";
this.#restartTimeout = null;
this.#pc = null;
this.#offerData = null;
this.#sessionUrl = null;
this.#queuedCandidates = [];
this.#start();
}
/**
* Close the publisher and all its resources.
*/
close = () => {
close() {
this.#state = "closed";
if (this.#pc !== null) {
@@ -66,7 +60,7 @@ class MediaMTXWebRTCPublisher {
if (this.#restartTimeout !== null) {
clearTimeout(this.#restartTimeout);
}
};
}
static #unquoteCredential(v) {
return JSON.parse(`"${v}"`);
@@ -123,17 +117,16 @@ class MediaMTXWebRTCPublisher {
candidatesByMedia[mid].push(candidate);
}
let frag =
"a=ice-ufrag:" + od.iceUfrag + "\r\n" + "a=ice-pwd:" + od.icePwd + "\r\n";
let frag = `a=ice-ufrag:${od.iceUfrag}\r\n` + `a=ice-pwd:${od.icePwd}\r\n`;
let mid = 0;
for (const media of od.medias) {
if (candidatesByMedia[mid] !== undefined) {
frag += "m=" + media + "\r\n" + "a=mid:" + mid + "\r\n";
frag += `m=${media}\r\n` + `a=mid:${mid}\r\n`;
for (const candidate of candidatesByMedia[mid]) {
frag += "a=" + candidate.candidate + "\r\n";
frag += `a=${candidate.candidate}\r\n`;
}
}
mid++;
@@ -273,6 +266,12 @@ class MediaMTXWebRTCPublisher {
return sections.join("m=");
}
#restart() {
this.#restartTimeout = null;
this.#state = "running";
this.#start();
}
#start() {
this.#requestICEServers()
.then((iceServers) => this.#setupPeerConnection(iceServers))
@@ -303,11 +302,10 @@ class MediaMTXWebRTCPublisher {
this.#queuedCandidates = [];
this.#state = "restarting";
this.#restartTimeout = window.setTimeout(() => {
this.#restartTimeout = null;
this.#state = "running";
this.#start();
}, this.#retryPause);
this.#restartTimeout = window.setTimeout(
() => this.#restart(),
MediaMTXWebRTCPublisher.#RETRY_PAUSE,
);
if (this.#conf.onError !== undefined) {
this.#conf.onError(`${err}, retrying in some seconds`);
@@ -329,9 +327,7 @@ class MediaMTXWebRTCPublisher {
#requestICEServers() {
return fetch(this.#conf.url, {
method: "OPTIONS",
headers: {
...this.#authHeader(),
},
headers: this.#authHeader(),
}).then((res) =>
MediaMTXWebRTCPublisher.#linkToIceServers(res.headers.get("Link")),
);
internal/servers/webrtc/read_index.html
+1
View File
@@ -47,6 +47,7 @@
<script>
const video = document.getElementById("video");
const message = document.getElementById("message");
let defaultControls = false;
let reader = null;
internal/servers/webrtc/reader.js
+19 -22
View File
@@ -29,14 +29,15 @@
/** WebRTC/WHEP reader. */
class MediaMTXWebRTCReader {
#retryPause;
static #RETRY_PAUSE = 2000;
#conf;
#state;
#restartTimeout;
#pc;
#offerData;
#sessionUrl;
#queuedCandidates;
#state = "getting_codecs";
#restartTimeout = null;
#pc = null;
#offerData = null;
#sessionUrl = null;
#queuedCandidates = [];
#nonAdvertisedCodecs;
/**
@@ -44,14 +45,7 @@ class MediaMTXWebRTCReader {
* @param {Conf} conf - configuration.
*/
constructor(conf) {
this.#retryPause = 2000;
this.#conf = conf;
this.#state = "getting_codecs";
this.#restartTimeout = null;
this.#pc = null;
this.#offerData = null;
this.#sessionUrl = null;
this.#queuedCandidates = [];
this.#getNonAdvertisedCodecs();
}
@@ -432,11 +426,10 @@ class MediaMTXWebRTCReader {
this.#queuedCandidates = [];
this.#state = "restarting";
this.#restartTimeout = window.setTimeout(() => {
this.#restartTimeout = null;
this.#state = "running";
this.#start();
}, this.#retryPause);
this.#restartTimeout = window.setTimeout(
() => this.#restart(),
MediaMTXWebRTCReader.#RETRY_PAUSE,
);
if (this.#conf.onError !== undefined) {
this.#conf.onError(`${err}, retrying in some seconds`);
@@ -450,6 +443,12 @@ class MediaMTXWebRTCReader {
}
}
#restart() {
this.#restartTimeout = null;
this.#state = "running";
this.#start();
}
#getNonAdvertisedCodecs() {
Promise.all(
[
@@ -504,9 +503,7 @@ class MediaMTXWebRTCReader {
#requestICEServers() {
return fetch(this.#conf.url, {
method: "OPTIONS",
headers: {
...this.#authHeader(),
},
headers: this.#authHeader(),
}).then((res) =>
MediaMTXWebRTCReader.#linkToIceServers(res.headers.get("Link")),
);
internal/servers/webrtc/server.go
+40 -41
View File
@@ -133,43 +133,40 @@ type serverAPISessionsKickReq struct {
res chan serverAPISessionsKickRes
}
type webRTCNewSessionRes struct {
type newSessionRes struct {
sx *session
answer []byte
errStatusCode int
err error
}
type webRTCNewSessionReq struct {
type newSessionReq struct {
pathName string
remoteAddr string
offer []byte
publish bool
httpRequest *http.Request
res chan webRTCNewSessionRes
res chan newSessionRes
}
type webRTCAddSessionCandidatesRes struct {
type addSessionCandidatesRes struct {
sx *session
answer *whip.SDPFragment
err error
}
type webRTCAddSessionCandidatesReq struct {
pathName string
type addSessionCandidatesReq struct {
secret uuid.UUID
fragment *whip.SDPFragment
res chan webRTCAddSessionCandidatesRes
res chan addSessionCandidatesRes
}
type webRTCDeleteSessionRes struct {
type deleteSessionRes struct {
err error
}
type webRTCDeleteSessionReq struct {
pathName string
secret uuid.UUID
res chan webRTCDeleteSessionRes
type deleteSessionReq struct {
secret uuid.UUID
res chan deleteSessionRes
}
type serverMetrics interface {
@@ -223,10 +220,10 @@ type Server struct {
sessionsBySecret map[uuid.UUID]*session
// in
chNewSession chan webRTCNewSessionReq
chNewSession chan newSessionReq
chCloseSession chan *session
chAddSessionCandidates chan webRTCAddSessionCandidatesReq
chDeleteSession chan webRTCDeleteSessionReq
chAddSessionCandidates chan addSessionCandidatesReq
chDeleteSession chan deleteSessionReq
chAPISessionsList chan serverAPISessionsListReq
chAPISessionsGet chan serverAPISessionsGetReq
chAPIConnsKick chan serverAPISessionsKickReq
@@ -243,10 +240,10 @@ func (s *Server) Initialize() error {
s.ctxCancel = ctxCancel
s.sessions = make(map[*session]struct{})
s.sessionsBySecret = make(map[uuid.UUID]*session)
s.chNewSession = make(chan webRTCNewSessionReq)
s.chNewSession = make(chan newSessionReq)
s.chCloseSession = make(chan *session)
s.chAddSessionCandidates = make(chan webRTCAddSessionCandidatesReq)
s.chDeleteSession = make(chan webRTCDeleteSessionReq)
s.chAddSessionCandidates = make(chan addSessionCandidatesReq)
s.chDeleteSession = make(chan deleteSessionReq)
s.chAPISessionsList = make(chan serverAPISessionsListReq)
s.chAPISessionsGet = make(chan serverAPISessionsGetReq)
s.chAPIConnsKick = make(chan serverAPISessionsKickReq)
@@ -309,7 +306,7 @@ func (s *Server) Initialize() error {
}
}
str := "listener opened on " + s.Address
str := "started with listeners on " + s.Address
if !s.Encryption {
str += " (TCP/HTTP)"
} else {
@@ -339,7 +336,7 @@ func (s *Server) Log(level logger.Level, format string, args ...any) {
// Close closes the server.
func (s *Server) Close() {
s.Log(logger.Info, "listener is closing")
s.Log(logger.Info, "closing")
if !interfaceIsEmpty(s.Metrics) {
s.Metrics.SetWebRTCServer(nil)
@@ -369,7 +366,11 @@ outer:
stunGatherTimeout: s.STUNGatherTimeout,
handshakeTimeout: s.HandshakeTimeout,
trackGatherTimeout: s.TrackGatherTimeout,
req: req,
pathName: req.pathName,
remoteAddr: req.remoteAddr,
offer: req.offer,
publish: req.publish,
httpRequest: req.httpRequest,
wg: &wg,
externalCmdPool: s.ExternalCmdPool,
pathManager: s.PathManager,
@@ -378,7 +379,7 @@ outer:
sx.initialize()
s.sessions[sx] = struct{}{}
s.sessionsBySecret[sx.secret] = sx
req.res <- webRTCNewSessionRes{sx: sx}
req.res <- newSessionRes{sx: sx}
case sx := <-s.chCloseSession:
delete(s.sessions, sx)
@@ -386,17 +387,17 @@ outer:
case req := <-s.chAddSessionCandidates:
sx, ok := s.sessionsBySecret[req.secret]
if !ok || sx.req.pathName != req.pathName {
req.res <- webRTCAddSessionCandidatesRes{err: ErrSessionNotFound}
if !ok {
req.res <- addSessionCandidatesRes{err: ErrSessionNotFound}
continue
}
req.res <- webRTCAddSessionCandidatesRes{sx: sx}
req.res <- addSessionCandidatesRes{sx: sx}
case req := <-s.chDeleteSession:
sx, ok := s.sessionsBySecret[req.secret]
if !ok || sx.req.pathName != req.pathName {
req.res <- webRTCDeleteSessionRes{err: ErrSessionNotFound}
if !ok {
req.res <- deleteSessionRes{err: ErrSessionNotFound}
continue
}
@@ -404,7 +405,7 @@ outer:
delete(s.sessionsBySecret, sx.secret)
sx.Close()
req.res <- webRTCDeleteSessionRes{}
req.res <- deleteSessionRes{}
case req := <-s.chAPISessionsList:
data := &defs.APIWebRTCSessionList{
@@ -505,17 +506,15 @@ func (s *Server) generateICEServers(clientConfig bool) ([]pwebrtc.ICEServer, err
}
// newSession is called by webRTCHTTPServer.
func (s *Server) newSession(req webRTCNewSessionReq) webRTCNewSessionRes {
req.res = make(chan webRTCNewSessionRes)
func (s *Server) newSession(req newSessionReq) newSessionRes {
req.res = make(chan newSessionRes)
select {
case s.chNewSession <- req:
res := <-req.res
return res.sx.new(req)
return <-req.res
case <-s.ctx.Done():
return webRTCNewSessionRes{
return newSessionRes{
errStatusCode: http.StatusInternalServerError,
err: fmt.Errorf("terminated"),
}
@@ -532,9 +531,9 @@ func (s *Server) closeSession(sx *session) {
// addSessionCandidates is called by webRTCHTTPServer.
func (s *Server) addSessionCandidates(
req webRTCAddSessionCandidatesReq,
) webRTCAddSessionCandidatesRes {
req.res = make(chan webRTCAddSessionCandidatesRes)
req addSessionCandidatesReq,
) addSessionCandidatesRes {
req.res = make(chan addSessionCandidatesRes)
select {
case s.chAddSessionCandidates <- req:
res1 := <-req.res
@@ -545,13 +544,13 @@ func (s *Server) addSessionCandidates(
return res1.sx.addCandidates(req)
case <-s.ctx.Done():
return webRTCAddSessionCandidatesRes{err: fmt.Errorf("terminated")}
return addSessionCandidatesRes{err: fmt.Errorf("terminated")}
}
}
// deleteSession is called by webRTCHTTPServer.
func (s *Server) deleteSession(req webRTCDeleteSessionReq) error {
req.res = make(chan webRTCDeleteSessionRes)
func (s *Server) deleteSession(req deleteSessionReq) error {
req.res = make(chan deleteSessionRes)
select {
case s.chDeleteSession <- req:
res := <-req.res
internal/servers/webrtc/server_test.go
+5 -5
View File
@@ -422,7 +422,7 @@ func TestServerPublish(t *testing.T) {
su, err := url.Parse("http://myuser:mypass@localhost:8886/teststream/whip?param=value")
require.NoError(t, err)
track := &webrtc.OutgoingTrack{
track := &webrtc.OutboundTrack{
Caps: pwebrtc.RTPCodecCapability{
MimeType: pwebrtc.MimeTypeH264,
ClockRate: 90000,
@@ -434,7 +434,7 @@ func TestServerPublish(t *testing.T) {
HTTPClient: hc,
URL: su,
Publish: true,
OutgoingTracks: []*webrtc.OutgoingTrack{track},
OutboundTracks: []*webrtc.OutboundTrack{track},
Log: test.NilLogger,
}
@@ -737,7 +737,7 @@ func TestServerRead(t *testing.T) {
done := make(chan struct{})
wc.IncomingTracks()[0].OnPacketRTP = func(pkt *rtp.Packet) {
wc.InboundTracks()[0].OnPacketRTP = func(pkt *rtp.Packet) {
select {
case <-done:
default:
@@ -990,7 +990,7 @@ func TestServerICERestart(t *testing.T) {
su, err := url.Parse("http://localhost:8886/teststream/whip")
require.NoError(t, err)
track := &webrtc.OutgoingTrack{
track := &webrtc.OutboundTrack{
Caps: pwebrtc.RTPCodecCapability{
MimeType: pwebrtc.MimeTypeH264,
ClockRate: 90000,
@@ -1002,7 +1002,7 @@ func TestServerICERestart(t *testing.T) {
HTTPClient: hc,
URL: su,
Publish: true,
OutgoingTracks: []*webrtc.OutgoingTrack{track},
OutboundTracks: []*webrtc.OutboundTrack{track},
Log: test.NilLogger,
}
internal/servers/webrtc/session.go
+74 -64
View File
@@ -218,6 +218,16 @@ func fullAnswerToSDPFragment(answerSDP string) (*whip.SDPFragment, error) {
return frag, nil
}
type initialRequestRes struct {
answer []byte
err error
errStatusCode int
}
type initialRequestReq struct {
res chan initialRequestRes
}
type sessionParent interface {
closeSession(sx *session)
generateICEServers(clientConfig bool) ([]pwebrtc.ICEServer, error)
@@ -235,7 +245,11 @@ type session struct {
stunGatherTimeout conf.Duration
handshakeTimeout conf.Duration
trackGatherTimeout conf.Duration
req webRTCNewSessionReq
pathName string
remoteAddr string
offer []byte
publish bool
httpRequest *http.Request
wg *sync.WaitGroup
externalCmdPool *externalcmd.Pool
pathManager serverPathManager
@@ -251,22 +265,19 @@ type session struct {
pc *webrtc.PeerConnection
user string
chNew chan webRTCNewSessionReq
chAddCandidates chan webRTCAddSessionCandidatesReq
chInitialRequest chan initialRequestReq
chAddCandidates chan addSessionCandidatesReq
}
func (s *session) initialize() {
ctx, ctxCancel := context.WithCancel(s.parentCtx)
s.ctx = ctx
s.ctxCancel = ctxCancel
s.ctx, s.ctxCancel = context.WithCancel(s.parentCtx)
s.created = time.Now()
s.uuid = uuid.New()
s.secret = uuid.New()
s.chNew = make(chan webRTCNewSessionReq)
s.chAddCandidates = make(chan webRTCAddSessionCandidatesReq)
s.chInitialRequest = make(chan initialRequestReq)
s.chAddCandidates = make(chan addSessionCandidatesReq)
s.Log(logger.Info, "created by %s", s.req.remoteAddr)
s.Log(logger.Info, "created by %s", s.remoteAddr)
s.wg.Add(1)
@@ -296,16 +307,17 @@ func (s *session) run() {
}
func (s *session) runInner() error {
var req initialRequestReq
select {
case <-s.chNew:
case req = <-s.chInitialRequest:
case <-s.ctx.Done():
return fmt.Errorf("terminated")
}
errStatusCode, err := s.runInner2()
errStatusCode, err := s.runInner2(&req)
if errStatusCode != 0 {
s.req.res <- webRTCNewSessionRes{
req.res <- initialRequestRes{
errStatusCode: errStatusCode,
err: err,
}
@@ -314,24 +326,24 @@ func (s *session) runInner() error {
return err
}
func (s *session) runInner2() (int, error) {
if s.req.publish {
return s.runPublish()
func (s *session) runInner2(req *initialRequestReq) (int, error) {
if s.publish {
return s.runPublish(req)
}
return s.runRead()
return s.runRead(req)
}
func (s *session) runPublish() (int, error) {
ip, _, _ := net.SplitHostPort(s.req.remoteAddr)
func (s *session) runPublish(req *initialRequestReq) (int, error) {
ip, _, _ := net.SplitHostPort(s.remoteAddr)
res1, err := s.pathManager.FindPathConf(defs.PathFindPathConfReq{
AccessRequest: defs.PathAccessRequest{
Name: s.req.pathName,
Query: s.req.httpRequest.URL.RawQuery,
Name: s.pathName,
Query: s.httpRequest.URL.RawQuery,
Publish: true,
Proto: auth.ProtocolWebRTC,
ID: &s.uuid,
Credentials: httpp.Credentials(s.req.httpRequest),
Credentials: httpp.Credentials(s.httpRequest),
IP: net.ParseIP(ip),
},
})
@@ -380,7 +392,7 @@ func (s *session) runPublish() (int, error) {
pc.Close()
}()
offer := whipOffer(s.req.offer)
offer := whipOffer(s.offer)
var sdp sdp.SessionDescription
err = sdp.Unmarshal([]byte(offer.SDP))
@@ -403,9 +415,11 @@ func (s *session) runPublish() (int, error) {
return http.StatusBadRequest, err
}
s.writeAnswer(answer)
req.res <- initialRequestRes{
answer: []byte(answer.SDP),
}
go s.readRemoteCandidates(pc)
go s.readRemoteCandidates(s.offer, pc)
err = pc.WaitUntilConnected(time.Duration(s.handshakeTimeout))
if err != nil {
@@ -416,7 +430,7 @@ func (s *session) runPublish() (int, error) {
s.pc = pc
s.mutex.Unlock()
err = pc.GatherIncomingTracks(time.Duration(s.trackGatherTimeout))
err = pc.GatherInboundTracks(time.Duration(s.trackGatherTimeout))
if err != nil {
return 0, err
}
@@ -435,8 +449,8 @@ func (s *session) runPublish() (int, error) {
ReplaceNTP: !res1.Conf.UseAbsoluteTimestamp,
ConfToCompare: res1.Conf,
AccessRequest: defs.PathAccessRequest{
Name: s.req.pathName,
Query: s.req.httpRequest.URL.RawQuery,
Name: s.pathName,
Query: s.httpRequest.URL.RawQuery,
Publish: true,
SkipAuth: true,
},
@@ -460,17 +474,17 @@ func (s *session) runPublish() (int, error) {
}
}
func (s *session) runRead() (int, error) {
ip, _, _ := net.SplitHostPort(s.req.remoteAddr)
func (s *session) runRead(req *initialRequestReq) (int, error) {
ip, _, _ := net.SplitHostPort(s.remoteAddr)
res, err := s.pathManager.AddReader(defs.PathAddReaderReq{
Author: s,
AccessRequest: defs.PathAccessRequest{
Name: s.req.pathName,
Query: s.req.httpRequest.URL.RawQuery,
Name: s.pathName,
Query: s.httpRequest.URL.RawQuery,
Proto: auth.ProtocolWebRTC,
ID: &s.uuid,
Credentials: httpp.Credentials(s.req.httpRequest),
Credentials: httpp.Credentials(s.httpRequest),
IP: net.ParseIP(ip),
},
})
@@ -534,16 +548,18 @@ func (s *session) runRead() (int, error) {
pc.Close()
}()
offer := whipOffer(s.req.offer)
offer := whipOffer(s.offer)
answer, err := pc.CreateFullAnswer(offer, false)
if err != nil {
return http.StatusBadRequest, err
}
s.writeAnswer(answer)
req.res <- initialRequestRes{
answer: []byte(answer.SDP),
}
go s.readRemoteCandidates(pc)
go s.readRemoteCandidates(s.offer, pc)
err = pc.WaitUntilConnected(time.Duration(s.handshakeTimeout))
if err != nil {
@@ -563,7 +579,7 @@ func (s *session) runRead() (int, error) {
Conf: res.Path.SafeConf(),
ExternalCmdEnv: res.Path.ExternalCmdEnv(),
Reader: *s.APIReaderDescribe(),
Query: s.req.httpRequest.URL.RawQuery,
Query: s.httpRequest.URL.RawQuery,
})
defer onUnreadHook()
@@ -586,15 +602,8 @@ func (s *session) runRead() (int, error) {
}
}
func (s *session) writeAnswer(answer *pwebrtc.SessionDescription) {
s.req.res <- webRTCNewSessionRes{
sx: s,
answer: []byte(answer.SDP),
}
}
func (s *session) readRemoteCandidates(pc *webrtc.PeerConnection) {
remoteUfrag := parseOfferUfrag(s.req.offer)
func (s *session) readRemoteCandidates(offer []byte, pc *webrtc.PeerConnection) {
remoteUfrag := parseOfferUfrag(offer)
for {
select {
@@ -604,18 +613,18 @@ func (s *session) readRemoteCandidates(pc *webrtc.PeerConnection) {
candidates, err := sdpFragmentToCandidates(req.fragment)
if err != nil {
req.res <- webRTCAddSessionCandidatesRes{err: err}
req.res <- addSessionCandidatesRes{err: err}
continue
}
// ICE restart: client sent new credentials
var answer *pwebrtc.SessionDescription
if ufrag != "" && ufrag != remoteUfrag {
sdp := replaceICECredentials(s.req.offer, ufrag, pwd)
sdp := replaceICECredentials(offer, ufrag, pwd)
answer, err = pc.CreateFullAnswer(whipOffer(sdp), true)
if err != nil {
req.res <- webRTCAddSessionCandidatesRes{err: err}
req.res <- addSessionCandidatesRes{err: err}
continue
}
}
@@ -628,7 +637,7 @@ func (s *session) readRemoteCandidates(pc *webrtc.PeerConnection) {
}
}
if addErr != nil {
req.res <- webRTCAddSessionCandidatesRes{err: addErr}
req.res <- addSessionCandidatesRes{err: addErr}
continue
}
@@ -636,14 +645,14 @@ func (s *session) readRemoteCandidates(pc *webrtc.PeerConnection) {
var frag *whip.SDPFragment
frag, err = fullAnswerToSDPFragment(answer.SDP)
if err != nil {
req.res <- webRTCAddSessionCandidatesRes{err: err}
req.res <- addSessionCandidatesRes{err: err}
continue
}
remoteUfrag = ufrag
req.res <- webRTCAddSessionCandidatesRes{answer: frag}
req.res <- addSessionCandidatesRes{answer: frag}
} else {
req.res <- webRTCAddSessionCandidatesRes{}
req.res <- addSessionCandidatesRes{}
}
case <-s.ctx.Done():
@@ -652,27 +661,28 @@ func (s *session) readRemoteCandidates(pc *webrtc.PeerConnection) {
}
}
// new is called by webRTCHTTPServer through Server.
func (s *session) new(req webRTCNewSessionReq) webRTCNewSessionRes {
func (s *session) initialRequest(req initialRequestReq) initialRequestRes {
req.res = make(chan initialRequestRes)
select {
case s.chNew <- req:
case s.chInitialRequest <- req:
return <-req.res
case <-s.ctx.Done():
return webRTCNewSessionRes{err: fmt.Errorf("terminated"), errStatusCode: http.StatusInternalServerError}
return initialRequestRes{err: fmt.Errorf("terminated"), errStatusCode: http.StatusInternalServerError}
}
}
// addCandidates is called by webRTCHTTPServer through Server.
func (s *session) addCandidates(
req webRTCAddSessionCandidatesReq,
) webRTCAddSessionCandidatesRes {
req addSessionCandidatesReq,
) addSessionCandidatesRes {
select {
case s.chAddCandidates <- req:
return <-req.res
case <-s.ctx.Done():
return webRTCAddSessionCandidatesRes{err: fmt.Errorf("terminated")}
return addSessionCandidatesRes{err: fmt.Errorf("terminated")}
}
}
@@ -731,18 +741,18 @@ func (s *session) apiItem() *defs.APIWebRTCSession {
return &defs.APIWebRTCSession{
ID: s.uuid,
Created: s.created,
RemoteAddr: s.req.remoteAddr,
RemoteAddr: s.remoteAddr,
PeerConnectionEstablished: peerConnectionEstablished,
LocalCandidate: localCandidate,
RemoteCandidate: remoteCandidate,
State: func() defs.APIWebRTCSessionState {
if s.req.publish {
if s.publish {
return defs.APIWebRTCSessionStatePublish
}
return defs.APIWebRTCSessionStateRead
}(),
Path: s.req.pathName,
Query: s.req.httpRequest.URL.RawQuery,
Path: s.pathName,
Query: s.httpRequest.URL.RawQuery,
User: s.user,
InboundBytes: bytesReceived,
InboundRTPPackets: rtpPacketsReceived,
internal/staticsources/webrtc/source_test.go
+3 -3
View File
@@ -26,7 +26,7 @@ func whipOffer(body []byte) *pwebrtc.SessionDescription {
}
func TestSource(t *testing.T) {
outgoingTracks := []*webrtc.OutgoingTrack{{
outboundTracks := []*webrtc.OutboundTrack{{
Caps: pwebrtc.RTPCodecCapability{
MimeType: "audio/opus",
ClockRate: 48000,
@@ -39,7 +39,7 @@ func TestSource(t *testing.T) {
LocalRandomUDP: true,
IPsFromInterfaces: true,
Publish: true,
OutgoingTracks: outgoingTracks,
OutboundTracks: outboundTracks,
Log: test.NilLogger,
}
err := pc.Start()
@@ -81,7 +81,7 @@ func TestSource(t *testing.T) {
err3 := pc.WaitUntilConnected(10 * time.Second)
require.NoError(t, err3)
err3 = outgoingTracks[0].WriteRTP(&rtp.Packet{
err3 = outboundTracks[0].WriteRTP(&rtp.Packet{
Header: rtp.Header{
Version: 2,
Marker: true,