diff --git a/go.mod b/go.mod index 79609685d..6f20faf77 100644 --- a/go.mod +++ b/go.mod @@ -14,13 +14,13 @@ require ( github.com/google/wire v0.5.0 github.com/gorilla/websocket v1.4.2 github.com/hashicorp/golang-lru v0.5.4 - github.com/livekit/protocol v0.11.8-0.20220104065946-2c4c8d7764ed + github.com/livekit/protocol v0.11.8-0.20220108052220-a7f937bb7bba github.com/magefile/mage v1.11.0 github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0 github.com/mitchellh/go-homedir v1.1.0 github.com/olekukonko/tablewriter v0.0.5 - github.com/pion/ice/v2 v2.1.14 - github.com/pion/interceptor v0.1.0 + github.com/pion/ice/v2 v2.1.18 + github.com/pion/interceptor v0.1.4 github.com/pion/logging v0.2.2 github.com/pion/rtcp v1.2.9 github.com/pion/rtp v1.7.4 @@ -28,7 +28,7 @@ require ( github.com/pion/stun v0.3.5 github.com/pion/transport v0.13.0 github.com/pion/turn/v2 v2.0.6 - github.com/pion/webrtc/v3 v3.1.10 + github.com/pion/webrtc/v3 v3.1.14-0.20220107154618-d4b645635c44 github.com/pkg/errors v0.9.1 github.com/prometheus/client_golang v1.11.0 github.com/rs/zerolog v1.26.0 @@ -58,10 +58,10 @@ require ( github.com/mattn/go-runewidth v0.0.9 // indirect github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect github.com/pion/datachannel v1.5.2 // indirect - github.com/pion/dtls/v2 v2.0.10 // indirect + github.com/pion/dtls/v2 v2.0.13 // indirect github.com/pion/mdns v0.0.5 // indirect github.com/pion/randutil v0.1.0 // indirect - github.com/pion/sctp v1.8.0 // indirect + github.com/pion/sctp v1.8.2 // indirect github.com/pion/srtp/v2 v2.0.5 // indirect github.com/pion/udp v0.1.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect @@ -71,9 +71,9 @@ require ( github.com/russross/blackfriday/v2 v2.1.0 // indirect go.uber.org/atomic v1.7.0 // indirect go.uber.org/multierr v1.6.0 // indirect - golang.org/x/crypto v0.0.0-20210921155107-089bfa567519 // indirect + golang.org/x/crypto v0.0.0-20211117183948-ae814b36b871 // indirect golang.org/x/mod v0.5.1 // indirect - golang.org/x/net v0.0.0-20211201190559-0a0e4e1bb54c // indirect + golang.org/x/net v0.0.0-20211216030914-fe4d6282115f // indirect golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359 // indirect golang.org/x/text v0.3.6 // indirect golang.org/x/tools v0.1.7 // indirect diff --git a/go.sum b/go.sum index 6499baf25..9fbb09ff7 100644 --- a/go.sum +++ b/go.sum @@ -132,8 +132,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY/mKrcvA= github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts= -github.com/livekit/protocol v0.11.8-0.20220104065946-2c4c8d7764ed h1:6vxJ62pwuhXtEjqvsANTIoEcTgHR9laMa9tR3Xr0fAM= -github.com/livekit/protocol v0.11.8-0.20220104065946-2c4c8d7764ed/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg= +github.com/livekit/protocol v0.11.8-0.20220108052220-a7f937bb7bba h1:JFvsAPoALVyLzxyOZbknHZXqkQ867iMUKduBcZiD0D8= +github.com/livekit/protocol v0.11.8-0.20220108052220-a7f937bb7bba/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg= github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls= github.com/magefile/mage v1.11.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A= github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0= @@ -159,24 +159,23 @@ github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N github.com/olekukonko/tablewriter v0.0.5/go.mod h1:hPp6KlRPjbx+hW8ykQs1w3UBbZlj6HuIJcUGPhkA7kY= github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk= -github.com/onsi/ginkgo v1.16.1/go.mod h1:CObGmKUOKaSC0RjmoAK7tKyn4Azo5P2IWuoMnvwxz1E= -github.com/onsi/ginkgo v1.16.4 h1:29JGrr5oVBm5ulCWet69zQkzWipVXIol6ygQUe/EzNc= github.com/onsi/ginkgo v1.16.4/go.mod h1:dX+/inL/fNMqNlz0e9LfyB9TswhZpCVdJM/Z6Vvnwo0= +github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE= +github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU= github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY= github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo= github.com/onsi/gomega v1.10.3/go.mod h1:V9xEwhxec5O8UDM77eCW8vLymOMltsqPVYWrpDsH8xc= -github.com/onsi/gomega v1.11.0/go.mod h1:azGKhqFUon9Vuj0YmTfLSmx0FUwqXYSTl5re8lQLTUg= -github.com/onsi/gomega v1.15.0 h1:WjP/FQ/sk43MRmnEcT+MlDw2TFvkrXlprrPST/IudjU= github.com/onsi/gomega v1.15.0/go.mod h1:cIuvLEne0aoVhAgh/O6ac0Op8WWw9H6eYCriF+tEHG0= +github.com/onsi/gomega v1.17.0 h1:9Luw4uT5HTjHTN8+aNcSThgH1vdXnmdJ8xIfZ4wyTRE= +github.com/onsi/gomega v1.17.0/go.mod h1:HnhC7FXeEQY45zxNK3PPoIUhzk/80Xly9PcubAlGdZY= github.com/pion/datachannel v1.5.2 h1:piB93s8LGmbECrpO84DnkIVWasRMk3IimbcXkTQLE6E= github.com/pion/datachannel v1.5.2/go.mod h1:FTGQWaHrdCwIJ1rw6xBIfZVkslikjShim5yr05XFuCQ= -github.com/pion/dtls/v2 v2.0.9/go.mod h1:O0Wr7si/Zj5/EBFlDzDd6UtVxx25CE1r7XM7BQKYQho= -github.com/pion/dtls/v2 v2.0.10 h1:wgys7gPR1NMbWjmjJ3CW7lkUGaun8djgH8nahpNLnxI= -github.com/pion/dtls/v2 v2.0.10/go.mod h1:00OxfeCRWHShcqT9jx8pKKmBWuTt0NCZoVPCaC4VKvU= -github.com/pion/ice/v2 v2.1.14 h1:nD9GZs3MiR1/dPa5EiMRMe8hLBG3/qqCdx/hTS2g8VE= -github.com/pion/ice/v2 v2.1.14/go.mod h1:ovgYHUmwYLlRvcCLI67PnQ5YGe+upXZbGgllBDG/ktU= -github.com/pion/interceptor v0.1.0 h1:SlXKaDlEvSl7cr4j8fJykzVz4UdH+7UDtcvx+u01wLU= -github.com/pion/interceptor v0.1.0/go.mod h1:j5NIl3tJJPB3u8+Z2Xz8MZs/VV6rc+If9mXEKNuFmEM= +github.com/pion/dtls/v2 v2.0.13 h1:toLgXzq42/MEmfgkXDfzdnwLHMi4tfycaQPGkv9tzRE= +github.com/pion/dtls/v2 v2.0.13/go.mod h1:OaE7eTM+ppaUhJ99OTO4aHl9uY6vPrT1gPY27uNTxRY= +github.com/pion/ice/v2 v2.1.18 h1:mDzd+iPKJmU30p4Kb+RPjK9olORLqJmQdiTUnVba50g= +github.com/pion/ice/v2 v2.1.18/go.mod h1:9jDr0iIUg8P6+0Jq8QJ/eFSkX3JnsPd293TjCdkfpTs= +github.com/pion/interceptor v0.1.4 h1:qL2xrdR6taLkVxEQj39btwEPRO3i9yd/olEw6+20dag= +github.com/pion/interceptor v0.1.4/go.mod h1:Lh3JSl/cbJ2wP8I3ccrjh1K/deRGRn3UlSPuOTiHb6U= github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY= github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms= github.com/pion/mdns v0.0.5 h1:Q2oj/JB3NqfzY9xGZ1fPzZzK7sDSD8rZPOvcIQ10BCw= @@ -184,33 +183,30 @@ github.com/pion/mdns v0.0.5/go.mod h1:UgssrvdD3mxpi8tMxAXbsppL3vJ4Jipw1mTCW+al01 github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA= github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8= github.com/pion/rtcp v1.2.6/go.mod h1:52rMNPWFsjr39z9B9MhnkqhPLoeHTv1aN63o/42bWE0= -github.com/pion/rtcp v1.2.8/go.mod h1:qVPhiCzAm4D/rxb6XzKeyZiQK69yJpbUDJSF7TgrqNo= github.com/pion/rtcp v1.2.9 h1:1ujStwg++IOLIEoOiIQ2s+qBuJ1VN81KW+9pMPsif+U= github.com/pion/rtcp v1.2.9/go.mod h1:qVPhiCzAm4D/rxb6XzKeyZiQK69yJpbUDJSF7TgrqNo= github.com/pion/rtp v1.7.0/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko= -github.com/pion/rtp v1.7.2/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko= github.com/pion/rtp v1.7.4 h1:4dMbjb1SuynU5OpA3kz1zHK+u+eOCQjW3MAeVHf1ODA= github.com/pion/rtp v1.7.4/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko= -github.com/pion/sctp v1.8.0 h1:6erMF2qmQwXr+0iB1lm0AUSmDr9LdmpaBzgSVAEgehw= github.com/pion/sctp v1.8.0/go.mod h1:xFe9cLMZ5Vj6eOzpyiKjT9SwGM4KpK/8Jbw5//jc+0s= +github.com/pion/sctp v1.8.2 h1:yBBCIrUMJ4yFICL3RIvR4eh/H2BTTvlligmSTy+3kiA= +github.com/pion/sctp v1.8.2/go.mod h1:xFe9cLMZ5Vj6eOzpyiKjT9SwGM4KpK/8Jbw5//jc+0s= github.com/pion/sdp/v3 v3.0.4 h1:2Kf+dgrzJflNCSw3TV5v2VLeI0s/qkzy2r5jlR0wzf8= github.com/pion/sdp/v3 v3.0.4/go.mod h1:bNiSknmJE0HYBprTHXKPQ3+JjacTv5uap92ueJZKsRk= github.com/pion/srtp/v2 v2.0.5 h1:ks3wcTvIUE/GHndO3FAvROQ9opy0uLELpwHJaQ1yqhQ= github.com/pion/srtp/v2 v2.0.5/go.mod h1:8k6AJlal740mrZ6WYxc4Dg6qDqqhxoRG2GSjlUhDF0A= github.com/pion/stun v0.3.5 h1:uLUCBCkQby4S1cf6CGuR9QrVOKcvUwFeemaC865QHDg= github.com/pion/stun v0.3.5/go.mod h1:gDMim+47EeEtfWogA37n6qXZS88L5V6LqFcf+DZA2UA= -github.com/pion/transport v0.10.1/go.mod h1:PBis1stIILMiis0PewDw91WJeLJkyIMcEk+DwKOzf4A= github.com/pion/transport v0.12.2/go.mod h1:N3+vZQD9HlDP5GWkZ85LohxNsDcNgofQmyL6ojX5d8Q= github.com/pion/transport v0.12.3/go.mod h1:OViWW9SP2peE/HbwBvARicmAVnesphkNkCVZIWJ6q9A= github.com/pion/transport v0.13.0 h1:KWTA5ZrQogizzYwPEciGtHPLwpAjE91FgXnyu+Hv2uY= github.com/pion/transport v0.13.0/go.mod h1:yxm9uXpK9bpBBWkITk13cLo1y5/ur5VQpG22ny6EP7g= -github.com/pion/turn/v2 v2.0.5/go.mod h1:APg43CFyt/14Uy7heYUOGWdkem/Wu4PhCO/bjyrTqMw= github.com/pion/turn/v2 v2.0.6 h1:AsXjSPR6Im15DMTB39NlfdTY9BQfieANPBjdg/aVNwY= github.com/pion/turn/v2 v2.0.6/go.mod h1:+y7xl719J8bAEVpSXBXvTxStjJv3hbz9YFflvkpcGPw= github.com/pion/udp v0.1.1 h1:8UAPvyqmsxK8oOjloDk4wUt63TzFe9WEJkg5lChlj7o= github.com/pion/udp v0.1.1/go.mod h1:6AFo+CMdKQm7UiA0eUPA8/eVCTx8jBIITLZHc9DWX5M= -github.com/pion/webrtc/v3 v3.1.10 h1:DO99F6/X1HrQho3LxTWHPjI3c388btBf56lR5UNRNNk= -github.com/pion/webrtc/v3 v3.1.10/go.mod h1:eL2HHZOvX+W+Q+lenuidTrWfMD9gG3aobtGcCqJ5G48= +github.com/pion/webrtc/v3 v3.1.14-0.20220107154618-d4b645635c44 h1:iyFRWBUrSxF8EHUtcvioKBzm1hOB/F15Zph93+7MWro= +github.com/pion/webrtc/v3 v3.1.14-0.20220107154618-d4b645635c44/go.mod h1:Vn1lK5z9AEmabct0TirkWdt67kpzd40KrpkxPIvQfgE= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= @@ -286,9 +282,8 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4= -golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4= -golang.org/x/crypto v0.0.0-20210921155107-089bfa567519 h1:7I4JAnoQBe7ZtJcBaYHi5UtiO8tQHbUSXxL+pnGRANg= -golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= +golang.org/x/crypto v0.0.0-20211117183948-ae814b36b871 h1:/pEO3GD/ABYAjuakUS6xSEmmlyVS4kxBNkeA9tLJiTI= +golang.org/x/crypto v0.0.0-20211117183948-ae814b36b871/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE= golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU= @@ -315,17 +310,15 @@ golang.org/x/net v0.0.0-20201006153459-a7d1128ccaa0/go.mod h1:sp8m0HH+o8qH0wwXwY golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20201026091529-146b70c837a4/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20201201195509-5d6afe98e0b7/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= -golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= -golang.org/x/net v0.0.0-20210331212208-0fccb6fa2b5c/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk= golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= -golang.org/x/net v0.0.0-20211005001312-d4b1ae081e3b/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= -golang.org/x/net v0.0.0-20211020060615-d418f374d309/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= -golang.org/x/net v0.0.0-20211201190559-0a0e4e1bb54c h1:WtYZ93XtWSO5KlOMgPZu7hXY9WhMZpprvlm5VwvAl8c= +golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20211201190559-0a0e4e1bb54c/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= +golang.org/x/net v0.0.0-20211216030914-fe4d6282115f h1:hEYJvxw1lSnWIl8X9ofsYMklzaDs90JI2az5YMd4fPM= +golang.org/x/net v0.0.0-20211216030914-fe4d6282115f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 05a660e09..bc3e06d69 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -51,6 +51,7 @@ type MediaTrack struct { receiver sfu.Receiver layerDimensions sync.Map // livekit.VideoQuality => *livekit.VideoLayer + layerSsrcs [livekit.VideoQuality_HIGH + 1]uint32 // track audio fraction lost statsLock sync.Mutex @@ -119,6 +120,12 @@ func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTra return t } +func (t *MediaTrack) TrySetSimulcastSSRC(layer uint8, ssrc uint32) { + if int(layer) < len(t.layerSsrcs) && t.layerSsrcs[layer] == 0 { + t.layerSsrcs[layer] = ssrc + } +} + func (t *MediaTrack) ID() livekit.TrackID { return livekit.TrackID(t.params.TrackInfo.Sid) } @@ -319,6 +326,13 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra t.simulcasted.TrySet(true) } + if t.IsSimulcast() { + layer := sfu.RidToLayer(track.RID()) + if int(layer) < len(t.layerSsrcs) { + t.layerSsrcs[layer] = uint32(track.SSRC()) + } + } + buff.Bind(receiver.GetParameters(), track.Codec().RTPCodecCapability, buffer.Options{ MaxBitRate: t.params.ReceiverConfig.maxBitrate, }) @@ -331,6 +345,9 @@ func (t *MediaTrack) ToProto() *livekit.TrackInfo { layers := make([]*livekit.VideoLayer, 0) t.layerDimensions.Range(func(_, val interface{}) bool { if layer, ok := val.(*livekit.VideoLayer); ok { + if int(layer.Quality) < len(t.layerSsrcs) { + layer.Ssrc = t.layerSsrcs[layer.Quality] + } layers = append(layers, layer) } return true diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 675c4fb2b..9df177162 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -523,3 +523,13 @@ func (t *MediaTrackSubscriptions) maybeNotifyNoSubscribers() { t.onNoSubscribers() } } + +func (t *MediaTrackSubscriptions) GetAllSubscriberIDs() []livekit.ParticipantID { + t.subscribedTracksMu.RLock() + defer t.subscribedTracksMu.RUnlock() + ids := make([]livekit.ParticipantID, 0, len(t.subscribedTracks)) + for id := range t.subscribedTracks { + ids = append(ids, id) + } + return ids +} diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 296f4d394..dcf3c48c2 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -50,6 +50,7 @@ type ParticipantParams struct { Hidden bool Recorder bool Logger logger.Logger + SimTracks map[uint32]SimulcastTrackInfo } type ParticipantImpl struct { @@ -95,7 +96,10 @@ type ParticipantImpl struct { onStateChange func(p types.Participant, oldState livekit.ParticipantInfo_State) onMetadataUpdate func(types.Participant) onDataPacket func(types.Participant, *livekit.DataPacket) - onClose func(types.Participant, map[livekit.TrackID]livekit.ParticipantID) + + migrateState atomic.Value // types.MigrateState + pendingOffer *webrtc.SessionDescription + onClose func(types.Participant, map[livekit.TrackID]livekit.ParticipantID) } func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { @@ -107,6 +111,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { disallowedSubscriptions: make(map[livekit.TrackID]livekit.ParticipantID), connectedAt: time.Now(), } + p.migrateState.Store(types.MigrateStateInit) p.state.Store(livekit.ParticipantInfo_JOINING) var err error @@ -123,6 +128,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { Telemetry: p.params.Telemetry, EnabledCodecs: p.params.EnabledCodecs, Logger: params.Logger, + SimTracks: params.SimTracks, }) if err != nil { return nil, err @@ -175,7 +181,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { return nil, err } } - primaryPC.OnICEConnectionStateChange(p.handlePrimaryICEStateChange) + primaryPC.OnConnectionStateChange(p.handlePrimaryStateChange) p.publisher.pc.OnTrack(p.onMediaTrack) p.publisher.pc.OnDataChannel(p.onDataChannel) @@ -282,6 +288,13 @@ func (p *ParticipantImpl) OnClose(callback func(types.Participant, map[livekit.T // HandleOffer an offer from remote participant, used when clients make the initial connection func (p *ParticipantImpl) HandleOffer(sdp webrtc.SessionDescription) (answer webrtc.SessionDescription, err error) { + p.lock.Lock() + if p.MigrateState() == types.MigrateStateInit { + p.pendingOffer = &sdp + p.lock.Unlock() + return + } + p.lock.Unlock() p.params.Logger.Debugw("answering pub offer", "state", p.State().String(), // "sdp", sdp.SDP, @@ -327,6 +340,10 @@ func (p *ParticipantImpl) HandleOffer(sdp webrtc.SessionDescription) (answer web return } +func (p *ParticipantImpl) AddMigratedTrack(cid string, ti *livekit.TrackInfo) { + p.UptrackManager.AddMigratedTrack(cid, ti) +} + // AddTrack is called when client intends to publish track. // records track details and lets client know it's ok to proceed func (p *ParticipantImpl) AddTrack(req *livekit.AddTrackRequest) { @@ -434,7 +451,40 @@ func (p *ParticipantImpl) Close() error { } func (p *ParticipantImpl) Negotiate() { - p.subscriber.Negotiate() + if p.MigrateState() != types.MigrateStateInit { + p.subscriber.Negotiate() + } +} + +func (p *ParticipantImpl) SetPreviousAnswer(previous *webrtc.SessionDescription) { + p.subscriber.SetPreviousAnswer(previous) +} + +func (p *ParticipantImpl) SetMigrateState(s types.MigrateState) { + p.lock.Lock() + preState := p.MigrateState() + if preState == types.MigrateComplete || preState == s { + p.lock.Unlock() + return + } + p.params.Logger.Debugw("SetMigrateState", "state", s) + var pendingOffer *webrtc.SessionDescription + p.migrateState.Store(s) + if s == types.MigrateStateSync { + if !p.UptrackManager.HasPendingMigratedTrack() { + p.migrateState.Store(types.MigrateComplete) + } + pendingOffer = p.pendingOffer + p.pendingOffer = nil + } + p.lock.Unlock() + if pendingOffer != nil { + p.HandleOffer(*pendingOffer) + } +} + +func (p *ParticipantImpl) MigrateState() types.MigrateState { + return p.migrateState.Load().(types.MigrateState) } // ICERestart restarts subscriber ICE connections @@ -774,6 +824,9 @@ func (p *ParticipantImpl) setupUptrackManager() { }) p.UptrackManager.OnTrackPublished(func(track types.PublishedTrack) { + if !p.UptrackManager.HasPendingMigratedTrack() { + p.SetMigrateState(types.MigrateComplete) + } if p.onTrackPublished != nil { p.onTrackPublished(p, track) } @@ -886,7 +939,7 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w return } - p.UptrackManager.MediaTrackReceived(track, rtpReceiver) + p.UptrackManager.MediaTrackReceived(track, rtpReceiver, p) } func (p *ParticipantImpl) onDataChannel(dc *webrtc.DataChannel) { @@ -931,11 +984,11 @@ func (p *ParticipantImpl) handleDataMessage(kind livekit.DataPacket_Kind, data [ } } -func (p *ParticipantImpl) handlePrimaryICEStateChange(state webrtc.ICEConnectionState) { - if state == webrtc.ICEConnectionStateConnected { +func (p *ParticipantImpl) handlePrimaryStateChange(state webrtc.PeerConnectionState) { + if state == webrtc.PeerConnectionStateConnected { prometheus.ServiceOperationCounter.WithLabelValues("ice_connection", "success", "").Add(1) p.updateState(livekit.ParticipantInfo_ACTIVE) - } else if state == webrtc.ICEConnectionStateFailed { + } else if state == webrtc.PeerConnectionStateFailed { // only close when failed, to allow clients opportunity to reconnect go func() { _ = p.Close() diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 8cb591523..315a1a3c7 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -55,7 +55,7 @@ func TestICEStateChange(t *testing.T) { p.onClose = func(participant types.Participant, disallowedSubscriptions map[livekit.TrackID]livekit.ParticipantID) { close(closeChan) } - p.handlePrimaryICEStateChange(webrtc.ICEConnectionStateFailed) + p.handlePrimaryStateChange(webrtc.PeerConnectionStateFailed) select { case <-closeChan: @@ -220,7 +220,7 @@ func TestMuteSetting(t *testing.T) { t.Run("can set mute when track is pending", func(t *testing.T) { p := newParticipantForTest("test") ti := &livekit.TrackInfo{Sid: "testTrack"} - p.UptrackManager.pendingTracks["cid"] = ti + p.UptrackManager.pendingTracks["cid"] = &pendingTrackInfo{TrackInfo: ti} p.SetTrackMuted(livekit.TrackID(ti.Sid), true, false) require.True(t, ti.Muted) diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index a50224493..0886cb8b3 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -250,6 +250,8 @@ func (r *Room) Join(participant types.Participant, opts *ParticipantOptions, ice return err } + participant.SetMigrateState(types.MigrateComplete) + if participant.SubscriberAsPrimary() { // initiates sub connection as primary participant.Negotiate() @@ -377,6 +379,10 @@ func (r *Room) UpdateSubscriptions( return nil } +func (r *Room) SyncState(participant types.Participant, state *livekit.SyncState) error { + return nil +} + func (r *Room) UpdateSubscriptionPermissions(participant types.Participant, permissions *livekit.UpdateSubscriptionPermissions) error { return participant.UpdateSubscriptionPermissions(permissions, r.GetParticipantBySid) } diff --git a/pkg/rtc/signalhandler.go b/pkg/rtc/signalhandler.go index d6d556a61..2210ae980 100644 --- a/pkg/rtc/signalhandler.go +++ b/pkg/rtc/signalhandler.go @@ -85,6 +85,12 @@ func HandleParticipantSignal(room types.Room, participant types.Participant, req pLogger.Warnw("could not update subscription permissions", err, "permissions", msg.SubscriptionPermissions) } + case *livekit.SignalRequest_SyncState: + err := room.SyncState(participant, msg.SyncState) + if err != nil { + pLogger.Warnw("could not sync subscribe state", err, + "state", msg.SyncState) + } } return nil } diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index c6814960d..7577d2bcd 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -1,6 +1,8 @@ package rtc import ( + "errors" + "strings" "sync" "time" @@ -30,6 +32,11 @@ const ( negotiationRetry ) +type SimulcastTrackInfo struct { + Mid string + Rid string +} + // PCTransport is a wrapper around PeerConnection, with some helper methods type PCTransport struct { pc *webrtc.PeerConnection @@ -46,6 +53,8 @@ type PCTransport struct { streamAllocator *sfu.StreamAllocator logger logger.Logger + + previousAnswer *webrtc.SessionDescription } type TransportParams struct { @@ -57,6 +66,7 @@ type TransportParams struct { Telemetry telemetry.TelemetryService EnabledCodecs []*livekit.Codec Logger logger.Logger + SimTracks map[uint32]SimulcastTrackInfo } // LK-TODO-SSBWE func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimator cc.BandwidthEstimator)) (*webrtc.PeerConnection, *webrtc.MediaEngine, error) { @@ -115,6 +125,14 @@ func newPeerConnection(params TransportParams) (*webrtc.PeerConnection, *webrtc. */ } } + if len(params.SimTracks) > 0 { + f, err := NewUnhandleSimulcastInterceptorFactory(UnhandleSimulcastTracks(params.SimTracks)) + if err != nil { + params.Logger.Errorw("NewUnhandleSimulcastInterceptorFactory failed", err) + } else { + ir.Add(f) + } + } api := webrtc.NewAPI( webrtc.WithMediaEngine(me), webrtc.WithSettingEngine(se), @@ -181,6 +199,8 @@ func (t *PCTransport) AddICECandidate(candidate webrtc.ICECandidateInit) error { return nil } + t.logger.Debugw("add candidate ", "candidate", candidate.Candidate) + return t.pc.AddICECandidate(candidate) } @@ -284,6 +304,14 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error { return nil } + if t.previousAnswer != nil { + t.previousAnswer = nil + if options == nil { + options = &webrtc.OfferOptions{} + } + options.ICERestart = true + } + offer, err := t.pc.CreateOffer(options) if err != nil { prometheus.ServiceOperationCounter.WithLabelValues("offer", "error", "create").Add(1) @@ -306,6 +334,101 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error { return nil } +func (t *PCTransport) preparePC(previousAnswer webrtc.SessionDescription) error { + // sticky data channel to first m-lines, if someday we don't send sdp without media streams to + // client's subscribe pc after joining, should change this step + parsed, err := previousAnswer.Unmarshal() + if err != nil { + return err + } + fp, fpHahs, err := extractFingerprint(parsed) + if err != nil { + return err + } + + // for pion generate unmatched sdp, it always append data channel to last m-lines, + // that is not consistent with our subscribe offer which data channel is first m-lines, + // so use a dumb pc to negotiate sdp with only data channel then the data channel will + // sticky to first m-lines(subsequent sdp negotiation will keep m-lines's sequence) + offer, err := t.pc.CreateOffer(nil) + if err != nil { + return err + } + t.pc.SetLocalDescription(offer) + + pc2, err := webrtc.NewPeerConnection(webrtc.Configuration{}) + if err != nil { + return err + } + defer pc2.Close() + + pc2.SetRemoteDescription(offer) + ans, err := pc2.CreateAnswer(nil) + if err != nil { + return err + } + + // replace client's fingerprint into dump pc's answer, for pion's dtls process, it will + // keep the firgerprint at first call of SetRemoteDescription, if dumb pc and client pc use + // different fingerprint, that will cause pion denied dtls data after handshake with client + // complete (can't pass fingerprint change). + // in this step, we don't established connection with dump pc(no candidate swap), just use + // sdp negotiation to sticky data channel and keep client's fingerprint + parsedAns, _ := ans.Unmarshal() + fpLine := fpHahs + " " + fp + replaceFP := func(attrs []sdp.Attribute, fpLine string) { + for k := range attrs { + if attrs[k].Key == "fingerprint" { + attrs[k].Value = fpLine + } + } + } + replaceFP(parsedAns.Attributes, fpLine) + for _, m := range parsedAns.MediaDescriptions { + replaceFP(m.Attributes, fpLine) + } + bytes, err := parsedAns.Marshal() + if err != nil { + return err + } + ans.SDP = string(bytes) + + return t.pc.SetRemoteDescription(ans) +} + +func (t *PCTransport) initPCWithPreviousAnswer(previousOffer webrtc.SessionDescription) error { + if err := t.preparePC(previousOffer); err != nil { + return err + } + + parsed, err := previousOffer.Unmarshal() + if err != nil { + return err + } + for _, m := range parsed.MediaDescriptions { + var codecType webrtc.RTPCodecType + switch m.MediaName.Media { + case "video": + codecType = webrtc.RTPCodecTypeVideo + case "audio": + codecType = webrtc.RTPCodecTypeAudio + default: + continue + } + tr, err := t.pc.AddTransceiverFromKind(codecType, webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionRecvonly}) + if err != nil { + return err + } + tr.Stop() + mid := getMidValue(m) + if mid == "" { + return errors.New("mid value not found") + } + tr.SetMid(mid) + } + return nil +} + func (t *PCTransport) OnStreamStateChange(f func(update *sfu.StreamStateUpdate) error) { if t.streamAllocator == nil { return @@ -332,3 +455,51 @@ func (t *PCTransport) RemoveTrack(subTrack types.SubscribedTrack) { t.streamAllocator.RemoveTrack(subTrack.DownTrack()) } + +func (t *PCTransport) SetPreviousAnswer(offer *webrtc.SessionDescription) { + t.lock.Lock() + defer t.lock.Unlock() + if t.pc.RemoteDescription() == nil && t.previousAnswer == nil { + t.previousAnswer = offer + t.initPCWithPreviousAnswer(*t.previousAnswer) + } +} + +func getMidValue(media *sdp.MediaDescription) string { + for _, attr := range media.Attributes { + if attr.Key == "mid" { + return attr.Value + } + } + return "" +} + +func extractFingerprint(desc *sdp.SessionDescription) (string, string, error) { + fingerprints := []string{} + + if fingerprint, haveFingerprint := desc.Attribute("fingerprint"); haveFingerprint { + fingerprints = append(fingerprints, fingerprint) + } + + for _, m := range desc.MediaDescriptions { + if fingerprint, haveFingerprint := m.Attribute("fingerprint"); haveFingerprint { + fingerprints = append(fingerprints, fingerprint) + } + } + + if len(fingerprints) < 1 { + return "", "", webrtc.ErrSessionDescriptionNoFingerprint + } + + for _, m := range fingerprints { + if m != fingerprints[0] { + return "", "", webrtc.ErrSessionDescriptionConflictingFingerprints + } + } + + parts := strings.Split(fingerprints[0], " ") + if len(parts) != 2 { + return "", "", webrtc.ErrSessionDescriptionInvalidFingerprint + } + return parts[1], parts[0], nil +} diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index cdb312bb1..6feb908ec 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -24,6 +24,14 @@ type AddSubscriberParams struct { TrackIDs []livekit.TrackID } +type MigrateState int32 + +const ( + MigrateStateInit MigrateState = iota + MigrateStateSync + MigrateComplete +) + //counterfeiter:generate . Participant type Participant interface { ID() livekit.ParticipantID @@ -40,8 +48,12 @@ type Participant interface { SubscriberMediaEngine() *webrtc.MediaEngine Negotiate() ICERestart() error + SetPreviousAnswer(previous *webrtc.SessionDescription) + SetMigrateState(s MigrateState) + MigrateState() MigrateState AddTrack(req *livekit.AddTrackRequest) + AddMigratedTrack(cid string, ti *livekit.TrackInfo) GetPublishedTrack(sid livekit.TrackID) PublishedTrack GetPublishedTracks() []PublishedTrack GetSubscribedTrack(sid livekit.TrackID) SubscribedTrack @@ -108,6 +120,7 @@ type Room interface { Name() livekit.RoomName UpdateSubscriptions(participant Participant, trackIDs []livekit.TrackID, participantTracks []*livekit.ParticipantTracks, subscribe bool) error UpdateSubscriptionPermissions(participant Participant, permissions *livekit.UpdateSubscriptionPermissions) error + SyncState(participant Participant, state *livekit.SyncState) error UpdateVideoLayers(participant Participant, updateVideoLayers *livekit.UpdateVideoLayers) error } @@ -133,6 +146,7 @@ type MediaTrack interface { AddSubscriber(participant Participant) error RemoveSubscriber(participantID livekit.ParticipantID) IsSubscriber(subID livekit.ParticipantID) bool + GetAllSubscriberIDs() []livekit.ParticipantID RemoveAllSubscribers() RevokeDisallowedSubscribers(allowedSubscriberIDs []livekit.ParticipantID) []livekit.ParticipantID diff --git a/pkg/rtc/types/protocol_version.go b/pkg/rtc/types/protocol_version.go index e87af33f3..b12711a66 100644 --- a/pkg/rtc/types/protocol_version.go +++ b/pkg/rtc/types/protocol_version.go @@ -35,3 +35,7 @@ func (v ProtocolVersion) SupportsTransceiverReuse() bool { func (v ProtocolVersion) SupportsConnectionQuality() bool { return v > 4 } + +func (v ProtocolVersion) SupportsSessionMigrate() bool { + return v > 5 +} diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index f818f7f0c..f0824b835 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -20,6 +20,16 @@ type FakeMediaTrack struct { addSubscriberReturnsOnCall map[int]struct { result1 error } + GetAllSubscriberIDsStub func() []livekit.ParticipantID + getAllSubscriberIDsMutex sync.RWMutex + getAllSubscriberIDsArgsForCall []struct { + } + getAllSubscriberIDsReturns struct { + result1 []livekit.ParticipantID + } + getAllSubscriberIDsReturnsOnCall map[int]struct { + result1 []livekit.ParticipantID + } GetQualityForDimensionStub func(uint32, uint32) livekit.VideoQuality getQualityForDimensionMutex sync.RWMutex getQualityForDimensionArgsForCall []struct { @@ -246,6 +256,59 @@ func (fake *FakeMediaTrack) AddSubscriberReturnsOnCall(i int, result1 error) { }{result1} } +func (fake *FakeMediaTrack) GetAllSubscriberIDs() []livekit.ParticipantID { + fake.getAllSubscriberIDsMutex.Lock() + ret, specificReturn := fake.getAllSubscriberIDsReturnsOnCall[len(fake.getAllSubscriberIDsArgsForCall)] + fake.getAllSubscriberIDsArgsForCall = append(fake.getAllSubscriberIDsArgsForCall, struct { + }{}) + stub := fake.GetAllSubscriberIDsStub + fakeReturns := fake.getAllSubscriberIDsReturns + fake.recordInvocation("GetAllSubscriberIDs", []interface{}{}) + fake.getAllSubscriberIDsMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeMediaTrack) GetAllSubscriberIDsCallCount() int { + fake.getAllSubscriberIDsMutex.RLock() + defer fake.getAllSubscriberIDsMutex.RUnlock() + return len(fake.getAllSubscriberIDsArgsForCall) +} + +func (fake *FakeMediaTrack) GetAllSubscriberIDsCalls(stub func() []livekit.ParticipantID) { + fake.getAllSubscriberIDsMutex.Lock() + defer fake.getAllSubscriberIDsMutex.Unlock() + fake.GetAllSubscriberIDsStub = stub +} + +func (fake *FakeMediaTrack) GetAllSubscriberIDsReturns(result1 []livekit.ParticipantID) { + fake.getAllSubscriberIDsMutex.Lock() + defer fake.getAllSubscriberIDsMutex.Unlock() + fake.GetAllSubscriberIDsStub = nil + fake.getAllSubscriberIDsReturns = struct { + result1 []livekit.ParticipantID + }{result1} +} + +func (fake *FakeMediaTrack) GetAllSubscriberIDsReturnsOnCall(i int, result1 []livekit.ParticipantID) { + fake.getAllSubscriberIDsMutex.Lock() + defer fake.getAllSubscriberIDsMutex.Unlock() + fake.GetAllSubscriberIDsStub = nil + if fake.getAllSubscriberIDsReturnsOnCall == nil { + fake.getAllSubscriberIDsReturnsOnCall = make(map[int]struct { + result1 []livekit.ParticipantID + }) + } + fake.getAllSubscriberIDsReturnsOnCall[i] = struct { + result1 []livekit.ParticipantID + }{result1} +} + func (fake *FakeMediaTrack) GetQualityForDimension(arg1 uint32, arg2 uint32) livekit.VideoQuality { fake.getQualityForDimensionMutex.Lock() ret, specificReturn := fake.getQualityForDimensionReturnsOnCall[len(fake.getQualityForDimensionArgsForCall)] @@ -1141,6 +1204,8 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { defer fake.invocationsMutex.RUnlock() fake.addSubscriberMutex.RLock() defer fake.addSubscriberMutex.RUnlock() + fake.getAllSubscriberIDsMutex.RLock() + defer fake.getAllSubscriberIDsMutex.RUnlock() fake.getQualityForDimensionMutex.RLock() defer fake.getQualityForDimensionMutex.RUnlock() fake.iDMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 790cbb0b8..1f1d621e4 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -24,6 +24,12 @@ type FakeParticipant struct { addICECandidateReturnsOnCall map[int]struct { result1 error } + AddMigratedTrackStub func(string, *livekit.TrackInfo) + addMigratedTrackMutex sync.RWMutex + addMigratedTrackArgsForCall []struct { + arg1 string + arg2 *livekit.TrackInfo + } AddSubscribedTrackStub func(types.SubscribedTrack) addSubscribedTrackMutex sync.RWMutex addSubscribedTrackArgsForCall []struct { @@ -287,6 +293,16 @@ type FakeParticipant struct { isSubscribedToReturnsOnCall map[int]struct { result1 bool } + MigrateStateStub func() types.MigrateState + migrateStateMutex sync.RWMutex + migrateStateArgsForCall []struct { + } + migrateStateReturns struct { + result1 types.MigrateState + } + migrateStateReturnsOnCall map[int]struct { + result1 types.MigrateState + } NegotiateStub func() negotiateMutex sync.RWMutex negotiateArgsForCall []struct { @@ -416,11 +432,21 @@ type FakeParticipant struct { setMetadataArgsForCall []struct { arg1 string } + SetMigrateStateStub func(types.MigrateState) + setMigrateStateMutex sync.RWMutex + setMigrateStateArgsForCall []struct { + arg1 types.MigrateState + } SetPermissionStub func(*livekit.ParticipantPermission) setPermissionMutex sync.RWMutex setPermissionArgsForCall []struct { arg1 *livekit.ParticipantPermission } + SetPreviousAnswerStub func(*webrtc.SessionDescription) + setPreviousAnswerMutex sync.RWMutex + setPreviousAnswerArgsForCall []struct { + arg1 *webrtc.SessionDescription + } SetResponseSinkStub func(routing.MessageSink) setResponseSinkMutex sync.RWMutex setResponseSinkArgsForCall []struct { @@ -609,6 +635,39 @@ func (fake *FakeParticipant) AddICECandidateReturnsOnCall(i int, result1 error) }{result1} } +func (fake *FakeParticipant) AddMigratedTrack(arg1 string, arg2 *livekit.TrackInfo) { + fake.addMigratedTrackMutex.Lock() + fake.addMigratedTrackArgsForCall = append(fake.addMigratedTrackArgsForCall, struct { + arg1 string + arg2 *livekit.TrackInfo + }{arg1, arg2}) + stub := fake.AddMigratedTrackStub + fake.recordInvocation("AddMigratedTrack", []interface{}{arg1, arg2}) + fake.addMigratedTrackMutex.Unlock() + if stub != nil { + fake.AddMigratedTrackStub(arg1, arg2) + } +} + +func (fake *FakeParticipant) AddMigratedTrackCallCount() int { + fake.addMigratedTrackMutex.RLock() + defer fake.addMigratedTrackMutex.RUnlock() + return len(fake.addMigratedTrackArgsForCall) +} + +func (fake *FakeParticipant) AddMigratedTrackCalls(stub func(string, *livekit.TrackInfo)) { + fake.addMigratedTrackMutex.Lock() + defer fake.addMigratedTrackMutex.Unlock() + fake.AddMigratedTrackStub = stub +} + +func (fake *FakeParticipant) AddMigratedTrackArgsForCall(i int) (string, *livekit.TrackInfo) { + fake.addMigratedTrackMutex.RLock() + defer fake.addMigratedTrackMutex.RUnlock() + argsForCall := fake.addMigratedTrackArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + func (fake *FakeParticipant) AddSubscribedTrack(arg1 types.SubscribedTrack) { fake.addSubscribedTrackMutex.Lock() fake.addSubscribedTrackArgsForCall = append(fake.addSubscribedTrackArgsForCall, struct { @@ -2003,6 +2062,59 @@ func (fake *FakeParticipant) IsSubscribedToReturnsOnCall(i int, result1 bool) { }{result1} } +func (fake *FakeParticipant) MigrateState() types.MigrateState { + fake.migrateStateMutex.Lock() + ret, specificReturn := fake.migrateStateReturnsOnCall[len(fake.migrateStateArgsForCall)] + fake.migrateStateArgsForCall = append(fake.migrateStateArgsForCall, struct { + }{}) + stub := fake.MigrateStateStub + fakeReturns := fake.migrateStateReturns + fake.recordInvocation("MigrateState", []interface{}{}) + fake.migrateStateMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeParticipant) MigrateStateCallCount() int { + fake.migrateStateMutex.RLock() + defer fake.migrateStateMutex.RUnlock() + return len(fake.migrateStateArgsForCall) +} + +func (fake *FakeParticipant) MigrateStateCalls(stub func() types.MigrateState) { + fake.migrateStateMutex.Lock() + defer fake.migrateStateMutex.Unlock() + fake.MigrateStateStub = stub +} + +func (fake *FakeParticipant) MigrateStateReturns(result1 types.MigrateState) { + fake.migrateStateMutex.Lock() + defer fake.migrateStateMutex.Unlock() + fake.MigrateStateStub = nil + fake.migrateStateReturns = struct { + result1 types.MigrateState + }{result1} +} + +func (fake *FakeParticipant) MigrateStateReturnsOnCall(i int, result1 types.MigrateState) { + fake.migrateStateMutex.Lock() + defer fake.migrateStateMutex.Unlock() + fake.MigrateStateStub = nil + if fake.migrateStateReturnsOnCall == nil { + fake.migrateStateReturnsOnCall = make(map[int]struct { + result1 types.MigrateState + }) + } + fake.migrateStateReturnsOnCall[i] = struct { + result1 types.MigrateState + }{result1} +} + func (fake *FakeParticipant) Negotiate() { fake.negotiateMutex.Lock() fake.negotiateArgsForCall = append(fake.negotiateArgsForCall, struct { @@ -2758,6 +2870,38 @@ func (fake *FakeParticipant) SetMetadataArgsForCall(i int) string { return argsForCall.arg1 } +func (fake *FakeParticipant) SetMigrateState(arg1 types.MigrateState) { + fake.setMigrateStateMutex.Lock() + fake.setMigrateStateArgsForCall = append(fake.setMigrateStateArgsForCall, struct { + arg1 types.MigrateState + }{arg1}) + stub := fake.SetMigrateStateStub + fake.recordInvocation("SetMigrateState", []interface{}{arg1}) + fake.setMigrateStateMutex.Unlock() + if stub != nil { + fake.SetMigrateStateStub(arg1) + } +} + +func (fake *FakeParticipant) SetMigrateStateCallCount() int { + fake.setMigrateStateMutex.RLock() + defer fake.setMigrateStateMutex.RUnlock() + return len(fake.setMigrateStateArgsForCall) +} + +func (fake *FakeParticipant) SetMigrateStateCalls(stub func(types.MigrateState)) { + fake.setMigrateStateMutex.Lock() + defer fake.setMigrateStateMutex.Unlock() + fake.SetMigrateStateStub = stub +} + +func (fake *FakeParticipant) SetMigrateStateArgsForCall(i int) types.MigrateState { + fake.setMigrateStateMutex.RLock() + defer fake.setMigrateStateMutex.RUnlock() + argsForCall := fake.setMigrateStateArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeParticipant) SetPermission(arg1 *livekit.ParticipantPermission) { fake.setPermissionMutex.Lock() fake.setPermissionArgsForCall = append(fake.setPermissionArgsForCall, struct { @@ -2790,6 +2934,38 @@ func (fake *FakeParticipant) SetPermissionArgsForCall(i int) *livekit.Participan return argsForCall.arg1 } +func (fake *FakeParticipant) SetPreviousAnswer(arg1 *webrtc.SessionDescription) { + fake.setPreviousAnswerMutex.Lock() + fake.setPreviousAnswerArgsForCall = append(fake.setPreviousAnswerArgsForCall, struct { + arg1 *webrtc.SessionDescription + }{arg1}) + stub := fake.SetPreviousAnswerStub + fake.recordInvocation("SetPreviousAnswer", []interface{}{arg1}) + fake.setPreviousAnswerMutex.Unlock() + if stub != nil { + fake.SetPreviousAnswerStub(arg1) + } +} + +func (fake *FakeParticipant) SetPreviousAnswerCallCount() int { + fake.setPreviousAnswerMutex.RLock() + defer fake.setPreviousAnswerMutex.RUnlock() + return len(fake.setPreviousAnswerArgsForCall) +} + +func (fake *FakeParticipant) SetPreviousAnswerCalls(stub func(*webrtc.SessionDescription)) { + fake.setPreviousAnswerMutex.Lock() + defer fake.setPreviousAnswerMutex.Unlock() + fake.SetPreviousAnswerStub = stub +} + +func (fake *FakeParticipant) SetPreviousAnswerArgsForCall(i int) *webrtc.SessionDescription { + fake.setPreviousAnswerMutex.RLock() + defer fake.setPreviousAnswerMutex.RUnlock() + argsForCall := fake.setPreviousAnswerArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeParticipant) SetResponseSink(arg1 routing.MessageSink) { fake.setResponseSinkMutex.Lock() fake.setResponseSinkArgsForCall = append(fake.setResponseSinkArgsForCall, struct { @@ -3433,6 +3609,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} { defer fake.invocationsMutex.RUnlock() fake.addICECandidateMutex.RLock() defer fake.addICECandidateMutex.RUnlock() + fake.addMigratedTrackMutex.RLock() + defer fake.addMigratedTrackMutex.RUnlock() fake.addSubscribedTrackMutex.RLock() defer fake.addSubscribedTrackMutex.RUnlock() fake.addSubscriberMutex.RLock() @@ -3485,6 +3663,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} { defer fake.isRecorderMutex.RUnlock() fake.isSubscribedToMutex.RLock() defer fake.isSubscribedToMutex.RUnlock() + fake.migrateStateMutex.RLock() + defer fake.migrateStateMutex.RUnlock() fake.negotiateMutex.RLock() defer fake.negotiateMutex.RUnlock() fake.onCloseMutex.RLock() @@ -3519,8 +3699,12 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} { defer fake.sendSpeakerUpdateMutex.RUnlock() fake.setMetadataMutex.RLock() defer fake.setMetadataMutex.RUnlock() + fake.setMigrateStateMutex.RLock() + defer fake.setMigrateStateMutex.RUnlock() fake.setPermissionMutex.RLock() defer fake.setPermissionMutex.RUnlock() + fake.setPreviousAnswerMutex.RLock() + defer fake.setPreviousAnswerMutex.RUnlock() fake.setResponseSinkMutex.RLock() defer fake.setResponseSinkMutex.RUnlock() fake.setTrackMutedMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_published_track.go b/pkg/rtc/types/typesfakes/fake_published_track.go index 7a159a2d5..084d47362 100644 --- a/pkg/rtc/types/typesfakes/fake_published_track.go +++ b/pkg/rtc/types/typesfakes/fake_published_track.go @@ -26,6 +26,16 @@ type FakePublishedTrack struct { addSubscriberReturnsOnCall map[int]struct { result1 error } + GetAllSubscriberIDsStub func() []livekit.ParticipantID + getAllSubscriberIDsMutex sync.RWMutex + getAllSubscriberIDsArgsForCall []struct { + } + getAllSubscriberIDsReturns struct { + result1 []livekit.ParticipantID + } + getAllSubscriberIDsReturnsOnCall map[int]struct { + result1 []livekit.ParticipantID + } GetAudioLevelStub func() (uint8, bool) getAudioLevelMutex sync.RWMutex getAudioLevelArgsForCall []struct { @@ -358,6 +368,59 @@ func (fake *FakePublishedTrack) AddSubscriberReturnsOnCall(i int, result1 error) }{result1} } +func (fake *FakePublishedTrack) GetAllSubscriberIDs() []livekit.ParticipantID { + fake.getAllSubscriberIDsMutex.Lock() + ret, specificReturn := fake.getAllSubscriberIDsReturnsOnCall[len(fake.getAllSubscriberIDsArgsForCall)] + fake.getAllSubscriberIDsArgsForCall = append(fake.getAllSubscriberIDsArgsForCall, struct { + }{}) + stub := fake.GetAllSubscriberIDsStub + fakeReturns := fake.getAllSubscriberIDsReturns + fake.recordInvocation("GetAllSubscriberIDs", []interface{}{}) + fake.getAllSubscriberIDsMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakePublishedTrack) GetAllSubscriberIDsCallCount() int { + fake.getAllSubscriberIDsMutex.RLock() + defer fake.getAllSubscriberIDsMutex.RUnlock() + return len(fake.getAllSubscriberIDsArgsForCall) +} + +func (fake *FakePublishedTrack) GetAllSubscriberIDsCalls(stub func() []livekit.ParticipantID) { + fake.getAllSubscriberIDsMutex.Lock() + defer fake.getAllSubscriberIDsMutex.Unlock() + fake.GetAllSubscriberIDsStub = stub +} + +func (fake *FakePublishedTrack) GetAllSubscriberIDsReturns(result1 []livekit.ParticipantID) { + fake.getAllSubscriberIDsMutex.Lock() + defer fake.getAllSubscriberIDsMutex.Unlock() + fake.GetAllSubscriberIDsStub = nil + fake.getAllSubscriberIDsReturns = struct { + result1 []livekit.ParticipantID + }{result1} +} + +func (fake *FakePublishedTrack) GetAllSubscriberIDsReturnsOnCall(i int, result1 []livekit.ParticipantID) { + fake.getAllSubscriberIDsMutex.Lock() + defer fake.getAllSubscriberIDsMutex.Unlock() + fake.GetAllSubscriberIDsStub = nil + if fake.getAllSubscriberIDsReturnsOnCall == nil { + fake.getAllSubscriberIDsReturnsOnCall = make(map[int]struct { + result1 []livekit.ParticipantID + }) + } + fake.getAllSubscriberIDsReturnsOnCall[i] = struct { + result1 []livekit.ParticipantID + }{result1} +} + func (fake *FakePublishedTrack) GetAudioLevel() (uint8, bool) { fake.getAudioLevelMutex.Lock() ret, specificReturn := fake.getAudioLevelReturnsOnCall[len(fake.getAudioLevelArgsForCall)] @@ -1632,6 +1695,8 @@ func (fake *FakePublishedTrack) Invocations() map[string][][]interface{} { defer fake.addOnCloseMutex.RUnlock() fake.addSubscriberMutex.RLock() defer fake.addSubscriberMutex.RUnlock() + fake.getAllSubscriberIDsMutex.RLock() + defer fake.getAllSubscriberIDsMutex.RUnlock() fake.getAudioLevelMutex.RLock() defer fake.getAudioLevelMutex.RUnlock() fake.getConnectionScoreMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_room.go b/pkg/rtc/types/typesfakes/fake_room.go index 0309b83c0..56b7b0eaa 100644 --- a/pkg/rtc/types/typesfakes/fake_room.go +++ b/pkg/rtc/types/typesfakes/fake_room.go @@ -19,6 +19,18 @@ type FakeRoom struct { nameReturnsOnCall map[int]struct { result1 livekit.RoomName } + SyncStateStub func(types.Participant, *livekit.SyncState) error + syncStateMutex sync.RWMutex + syncStateArgsForCall []struct { + arg1 types.Participant + arg2 *livekit.SyncState + } + syncStateReturns struct { + result1 error + } + syncStateReturnsOnCall map[int]struct { + result1 error + } UpdateSubscriptionPermissionsStub func(types.Participant, *livekit.UpdateSubscriptionPermissions) error updateSubscriptionPermissionsMutex sync.RWMutex updateSubscriptionPermissionsArgsForCall []struct { @@ -114,6 +126,68 @@ func (fake *FakeRoom) NameReturnsOnCall(i int, result1 livekit.RoomName) { }{result1} } +func (fake *FakeRoom) SyncState(arg1 types.Participant, arg2 *livekit.SyncState) error { + fake.syncStateMutex.Lock() + ret, specificReturn := fake.syncStateReturnsOnCall[len(fake.syncStateArgsForCall)] + fake.syncStateArgsForCall = append(fake.syncStateArgsForCall, struct { + arg1 types.Participant + arg2 *livekit.SyncState + }{arg1, arg2}) + stub := fake.SyncStateStub + fakeReturns := fake.syncStateReturns + fake.recordInvocation("SyncState", []interface{}{arg1, arg2}) + fake.syncStateMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeRoom) SyncStateCallCount() int { + fake.syncStateMutex.RLock() + defer fake.syncStateMutex.RUnlock() + return len(fake.syncStateArgsForCall) +} + +func (fake *FakeRoom) SyncStateCalls(stub func(types.Participant, *livekit.SyncState) error) { + fake.syncStateMutex.Lock() + defer fake.syncStateMutex.Unlock() + fake.SyncStateStub = stub +} + +func (fake *FakeRoom) SyncStateArgsForCall(i int) (types.Participant, *livekit.SyncState) { + fake.syncStateMutex.RLock() + defer fake.syncStateMutex.RUnlock() + argsForCall := fake.syncStateArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeRoom) SyncStateReturns(result1 error) { + fake.syncStateMutex.Lock() + defer fake.syncStateMutex.Unlock() + fake.SyncStateStub = nil + fake.syncStateReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeRoom) SyncStateReturnsOnCall(i int, result1 error) { + fake.syncStateMutex.Lock() + defer fake.syncStateMutex.Unlock() + fake.SyncStateStub = nil + if fake.syncStateReturnsOnCall == nil { + fake.syncStateReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.syncStateReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeRoom) UpdateSubscriptionPermissions(arg1 types.Participant, arg2 *livekit.UpdateSubscriptionPermissions) error { fake.updateSubscriptionPermissionsMutex.Lock() ret, specificReturn := fake.updateSubscriptionPermissionsReturnsOnCall[len(fake.updateSubscriptionPermissionsArgsForCall)] @@ -317,6 +391,8 @@ func (fake *FakeRoom) Invocations() map[string][][]interface{} { defer fake.invocationsMutex.RUnlock() fake.nameMutex.RLock() defer fake.nameMutex.RUnlock() + fake.syncStateMutex.RLock() + defer fake.syncStateMutex.RUnlock() fake.updateSubscriptionPermissionsMutex.RLock() defer fake.updateSubscriptionPermissionsMutex.RUnlock() fake.updateSubscriptionsMutex.RLock() diff --git a/pkg/rtc/unhandlesimulcast.go b/pkg/rtc/unhandlesimulcast.go new file mode 100644 index 000000000..877f57b32 --- /dev/null +++ b/pkg/rtc/unhandlesimulcast.go @@ -0,0 +1,125 @@ +package rtc + +import ( + "github.com/pion/interceptor" + "github.com/pion/rtp" + "github.com/pion/sdp/v3" + "github.com/pion/webrtc/v3" +) + +const ( + simulcastProbeCount = 10 +) + +type UnhandleSimulcastOption func(r *UnhandleSimulcastInterceptor) error + +func UnhandleSimulcastTracks(tracks map[uint32]SimulcastTrackInfo) UnhandleSimulcastOption { + return func(r *UnhandleSimulcastInterceptor) error { + r.simTracks = tracks + return nil + } +} + +type UnhandleSimulcastInterceptorFactory struct { + opts []UnhandleSimulcastOption +} + +func (f *UnhandleSimulcastInterceptorFactory) NewInterceptor(id string) (interceptor.Interceptor, error) { + i := &UnhandleSimulcastInterceptor{simTracks: map[uint32]SimulcastTrackInfo{}} + for _, o := range f.opts { + if err := o(i); err != nil { + return nil, err + } + } + return i, nil +} + +func NewUnhandleSimulcastInterceptorFactory(opts ...UnhandleSimulcastOption) (*UnhandleSimulcastInterceptorFactory, error) { + return &UnhandleSimulcastInterceptorFactory{opts: opts}, nil +} + +type unhandleSimulcastRTPReader struct { + SimulcastTrackInfo + tryTimes int + reader interceptor.RTPReader + midExtensionID uint8 + streamIDExtensionID uint8 +} + +func (r *unhandleSimulcastRTPReader) Read(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) { + n, a, err := r.reader.Read(b, a) + if r.tryTimes < 0 || err != nil { + return n, a, err + } + + header := rtp.Header{} + hsize, err := header.Unmarshal(b[:n]) + if err != nil { + return n, a, nil + } + var mid, rid string + if payload := header.GetExtension(r.midExtensionID); payload != nil { + mid = string(payload) + } + + if payload := header.GetExtension(r.streamIDExtensionID); payload != nil { + rid = string(payload) + } + + if mid != "" && rid != "" { + r.tryTimes = -1 + return n, a, nil + } + + r.tryTimes-- + + if mid == "" { + header.SetExtension(r.midExtensionID, []byte(r.Mid)) + } + if rid == "" { + header.SetExtension(r.streamIDExtensionID, []byte(r.Rid)) + } + + hsize2 := header.MarshalSize() + + if hsize2-hsize+n > len(b) { // no enough buf to set extension + return n, a, nil + } + copy(b[hsize2:], b[hsize:n]) + header.MarshalTo(b) + return hsize2 - hsize + n, a, nil +} + +type UnhandleSimulcastInterceptor struct { + interceptor.NoOp + simTracks map[uint32]SimulcastTrackInfo +} + +func getHeaderExtensionID(extensions []interceptor.RTPHeaderExtension, extension webrtc.RTPHeaderExtensionCapability) int { + for _, h := range extensions { + if extension.URI == h.URI { + return h.ID + } + } + return 0 +} + +func (u *UnhandleSimulcastInterceptor) BindRemoteStream(info *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader { + if t, ok := u.simTracks[info.SSRC]; ok { + // if we support fec for simulcast streams at future, should get rsid extensions + midExtensionID := getHeaderExtensionID(info.RTPHeaderExtensions, webrtc.RTPHeaderExtensionCapability{sdp.SDESMidURI}) + streamIDExtensionID := getHeaderExtensionID(info.RTPHeaderExtensions, webrtc.RTPHeaderExtensionCapability{sdp.SDESRTPStreamIDURI}) + if midExtensionID == 0 || streamIDExtensionID == 0 { + return reader + } + + return &unhandleSimulcastRTPReader{ + SimulcastTrackInfo: t, + reader: reader, + tryTimes: simulcastProbeCount, + midExtensionID: uint8(midExtensionID), + streamIDExtensionID: uint8(streamIDExtensionID), + } + } + return reader +} diff --git a/pkg/rtc/uptrackmanager.go b/pkg/rtc/uptrackmanager.go index a89dd0733..cf5e3b15f 100644 --- a/pkg/rtc/uptrackmanager.go +++ b/pkg/rtc/uptrackmanager.go @@ -12,6 +12,7 @@ import ( "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/rtc/types" + "github.com/livekit/livekit-server/pkg/sfu" "github.com/livekit/livekit-server/pkg/sfu/twcc" "github.com/livekit/livekit-server/pkg/telemetry" ) @@ -26,6 +27,11 @@ type UptrackManagerParams struct { Logger logger.Logger } +type pendingTrackInfo struct { + *livekit.TrackInfo + migrated bool +} + type UptrackManager struct { params UptrackManagerParams rtcpCh chan []rtcp.Packet @@ -39,7 +45,7 @@ type UptrackManager struct { // publishedTracks that participant is publishing publishedTracks map[livekit.TrackID]types.PublishedTrack // client intended to publish, yet to be reconciled - pendingTracks map[string]*livekit.TrackInfo + pendingTracks map[string]*pendingTrackInfo // keeps track of subscriptions that are awaiting permissions subscriptionPermissions map[livekit.ParticipantID]*livekit.TrackPermission // subscriberID => *livekit.TrackPermission // keeps tracks of track specific subscribers who are awaiting permission @@ -60,7 +66,7 @@ func NewUptrackManager(params UptrackManagerParams) *UptrackManager { rtcpCh: make(chan []rtcp.Packet, 50), pliThrottle: newPLIThrottle(params.ThrottleConfig), publishedTracks: make(map[livekit.TrackID]types.PublishedTrack, 0), - pendingTracks: make(map[string]*livekit.TrackInfo), + pendingTracks: make(map[string]*pendingTrackInfo), pendingSubscriptions: make(map[livekit.TrackID][]livekit.ParticipantID), } } @@ -139,11 +145,17 @@ func (u *UptrackManager) AddTrack(req *livekit.AddTrackRequest) *livekit.TrackIn Source: req.Source, Layers: req.Layers, } - u.pendingTracks[req.Cid] = ti + u.pendingTracks[req.Cid] = &pendingTrackInfo{TrackInfo: ti} return ti } +func (u *UptrackManager) AddMigratedTrack(cid string, ti *livekit.TrackInfo) { + u.lock.Lock() + defer u.lock.Unlock() + u.pendingTracks[cid] = &pendingTrackInfo{ti, true} +} + // AddSubscriber subscribes op to all publishedTracks func (u *UptrackManager) AddSubscriber(sub types.Participant, params types.AddSubscriberParams) (int, error) { var tracks []types.PublishedTrack @@ -283,7 +295,7 @@ func (u *UptrackManager) GetDTX() bool { var trackInfo *livekit.TrackInfo for _, ti := range u.pendingTracks { if ti.Type == livekit.TrackType_AUDIO { - trackInfo = ti + trackInfo = ti.TrackInfo break } } @@ -350,7 +362,7 @@ func (u *UptrackManager) UpdateMediaLoss(nodeID string, trackID livekit.TrackID, } // when a new remoteTrack is created, creates a Track and adds it to room -func (u *UptrackManager) MediaTrackReceived(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) { +func (u *UptrackManager) MediaTrackReceived(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver, p *ParticipantImpl) { var newTrack bool // use existing mediatrack to handle simulcast @@ -363,7 +375,15 @@ func (u *UptrackManager) MediaTrackReceived(track *webrtc.TrackRemote, rtpReceiv return } + var mid string + for _, tr := range p.publisher.pc.GetTransceivers() { + if tr.Receiver() == rtpReceiver { + mid = tr.Mid() + break + } + } ti.MimeType = track.Codec().MimeType + ti.Mid = mid mt = NewMediaTrack(track, MediaTrackParams{ TrackInfo: ti, @@ -379,6 +399,12 @@ func (u *UptrackManager) MediaTrackReceived(track *webrtc.TrackRemote, rtpReceiv Logger: u.params.Logger, SubscriberConfig: u.params.Config.Subscriber, }) + for ssrc, t := range p.params.SimTracks { + if t.Mid != mid { + continue + } + mt.TrySetSimulcastSSRC(uint8(sfu.RidToLayer(t.Rid)), ssrc) + } mt.OnSubscribedMaxQualityChange(u.onSubscribedMaxQualityChange) // add to published and clean up pending @@ -404,6 +430,8 @@ func (u *UptrackManager) MediaTrackReceived(track *webrtc.TrackRemote, rtpReceiv if newTrack { u.handleTrackPublished(mt) + } else { + u.onTrackUpdated(mt, true) } } @@ -458,7 +486,7 @@ func (u *UptrackManager) getPendingTrack(clientId string, kind livekit.TrackType if trackInfo == nil { u.params.Logger.Errorw("track info not published prior to track", nil, "clientId", clientId) } - return signalCid, trackInfo + return signalCid, trackInfo.TrackInfo } func (u *UptrackManager) handleTrackPublished(track types.PublishedTrack) { @@ -716,3 +744,14 @@ func (u *UptrackManager) DebugInfo() map[string]interface{} { return info } + +func (u *UptrackManager) HasPendingMigratedTrack() bool { + u.lock.RLock() + defer u.lock.RUnlock() + for _, t := range u.pendingTracks { + if t.migrated { + return true + } + } + return false +} diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 46bb4957a..49138b325 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -200,7 +200,8 @@ func (r *RoomManager) StartSession(ctx context.Context, roomName livekit.RoomNam room.RemoveParticipant(participant.Identity()) } } else if pi.Reconnect { - // send leave request if participant is trying to reconnect but missing from the room + // send leave request if participant is trying to reconnect without keep subscribe state + // but missing from the room if err = responseSink.WriteMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Leave{ Leave: &livekit.LeaveRequest{ diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index f08e4eed1..b1bcb95b2 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -107,6 +107,7 @@ func (s *RTCService) validate(r *http.Request) (livekit.RoomName, routing.Partic Recorder: claims.Video.Recorder, Client: s.parseClientInfo(r.Form), } + if autoSubParam != "" { pi.AutoSubscribe = boolValue(autoSubParam) } diff --git a/pkg/sfu/buffer/bucket.go b/pkg/sfu/buffer/bucket.go index bc2222e34..e844c851f 100644 --- a/pkg/sfu/buffer/bucket.go +++ b/pkg/sfu/buffer/bucket.go @@ -2,6 +2,7 @@ package buffer import ( "encoding/binary" + "fmt" "math" ) @@ -94,7 +95,7 @@ func (b *Bucket) get(sn uint16) []byte { func (b *Bucket) set(sn uint16, pkt []byte) ([]byte, error) { if b.headSN-sn >= uint16(b.maxSteps+1) { - return nil, ErrPacketTooOld + return nil, fmt.Errorf("%w, headSN %d, sn %d", ErrPacketTooOld, b.headSN, sn) } pos := b.step - int(b.headSN-sn+1) if pos < 0 { diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index db9084e92..2b61fc83b 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -89,6 +89,17 @@ type WebRTCReceiver struct { lbThreshold int } +func RidToLayer(rid string) int32 { + switch rid { + case FullResolution: + return 2 + case HalfResolution: + return 1 + default: + return 0 + } +} + type ReceiverOpts func(w *WebRTCReceiver) *WebRTCReceiver // WithPliThrottle indicates minimum time(ms) between sending PLIs @@ -182,15 +193,7 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff return } - var layer int32 - switch track.RID() { - case FullResolution: - layer = 2 - case HalfResolution: - layer = 1 - default: - layer = 0 - } + layer := RidToLayer(track.RID()) w.upTrackMu.Lock() w.upTracks[layer] = track