mirror of
https://github.com/livekit/livekit.git
synced 2026-10-22 11:22:27 +00:00
receiver class, imported relevant SFU helpers
This commit is contained in:
David Zhao committed
1 parent
71d3a470f5
commit
6c039bc64f
12 files changed
+890
-25
No files matched your search
@@ -3,10 +3,12 @@ module github.com/livekit/livekit-server
|
||||
go 1.15
|
||||
|
||||
require (
|
||||
github.com/gammazero/deque v0.0.0-20201010052221-3932da5530cc
|
||||
github.com/golang/protobuf v1.4.3
|
||||
github.com/google/uuid v1.1.2
|
||||
github.com/google/wire v0.4.0
|
||||
github.com/pion/ion-sfu v1.0.27
|
||||
github.com/pion/rtcp v1.2.4
|
||||
github.com/pion/rtp v1.6.1
|
||||
github.com/pion/sdp/v3 v3.0.2
|
||||
github.com/pion/stun v0.3.5
|
||||
@@ -15,6 +17,5 @@ require (
|
||||
github.com/urfave/cli/v2 v2.2.0
|
||||
github.com/urfave/negroni v1.0.0
|
||||
go.uber.org/zap v1.16.0
|
||||
google.golang.org/protobuf v1.25.0 // indirect
|
||||
gopkg.in/yaml.v2 v2.3.0
|
||||
)
|
||||
@@ -171,6 +171,7 @@ 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/lucas-clemente/quic-go v0.18.0 h1:JhQDdqxdwdmGdKsKgXi1+coHRoGhvU6z0rNzOJqZ/4o=
|
||||
github.com/lucas-clemente/quic-go v0.18.0/go.mod h1:yXttHsSNxQi8AWijC/vLP+OJczXqzHSOcJrM5ITUlCg=
|
||||
github.com/lucas-clemente/quic-go v0.18.1 h1:DMR7guC0NtVS8zNZR3IO7NARZvZygkSC56GGtC6cyys=
|
||||
github.com/lucas-clemente/quic-go v0.18.1/go.mod h1:yXttHsSNxQi8AWijC/vLP+OJczXqzHSOcJrM5ITUlCg=
|
||||
github.com/lucsky/cuid v1.0.2 h1:z4XlExeoderxoPj2/dxKOyPxe9RCOu7yNq9/XWxIUMQ=
|
||||
github.com/lucsky/cuid v1.0.2/go.mod h1:QaaJqckboimOmhRSJXSx/+IT+VTfxfPGSo/6mfgUfmE=
|
||||
@@ -182,6 +183,7 @@ github.com/marten-seemann/qtls v0.10.0 h1:ECsuYUKalRL240rRD4Ri33ISb7kAQ3qGDlrrl5
|
||||
github.com/marten-seemann/qtls v0.10.0/go.mod h1:UvMd1oaYDACI99/oZUYLzMCkBXQVT0aGm99sJhbT8hs=
|
||||
github.com/marten-seemann/qtls-go1-15 v0.1.0 h1:i/YPXVxz8q9umso/5y474CNcHmTpA+5DH+mFPjx6PZg=
|
||||
github.com/marten-seemann/qtls-go1-15 v0.1.0/go.mod h1:GyFwywLKkRt+6mfU99csTEY1joMZz5vmB1WNZH3P81I=
|
||||
github.com/marten-seemann/qtls-go1-15 v0.1.1 h1:LIH6K34bPVttyXnUWixk0bzH6/N07VxbSabxn5A5gZQ=
|
||||
github.com/marten-seemann/qtls-go1-15 v0.1.1/go.mod h1:GyFwywLKkRt+6mfU99csTEY1joMZz5vmB1WNZH3P81I=
|
||||
github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
|
||||
github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
|
||||
@@ -217,8 +219,6 @@ github.com/pion/datachannel v1.4.21 h1:3ZvhNyfmxsAqltQrApLPQMhSFNA+aT87RqyCq4OXm
|
||||
github.com/pion/datachannel v1.4.21/go.mod h1:oiNyP4gHx2DIwRzX/MFyH0Rz/Gz05OgBlayAI2hAWjg=
|
||||
github.com/pion/dtls/v2 v2.0.3 h1:3qQ0s4+TXD00rsllL8g8KQcxAs+Y/Z6oz618RXX6p14=
|
||||
github.com/pion/dtls/v2 v2.0.3/go.mod h1:TUjyL8bf8LH95h81Xj7kATmzMRt29F/4lxpIPj2Xe4Y=
|
||||
github.com/pion/ice/v2 v2.0.7 h1:3Iv+EfthCdPbJCsPbhnL4xRzolR1oh9mJTuUsxyxPJI=
|
||||
github.com/pion/ice/v2 v2.0.7/go.mod h1:0C4VBZxkO3A7Qj5ZEgZut8OptbJFSbuGNyPyJMftuHA=
|
||||
github.com/pion/ice/v2 v2.0.9 h1:oHbiN6Q9tgb8Gfu3I4cbr5mHRE1uqiuFABQ8CbWjIyk=
|
||||
github.com/pion/ice/v2 v2.0.9/go.mod h1:NK+o39ynb+N1YSj9fPgWs3vjVcrsWw0KCr/311MqVq8=
|
||||
github.com/pion/ion-sfu v1.0.27 h1:VR0LNI6Gno5rKUuhUoiESLGkkyd4u1QQtzg0SSUt5F8=
|
||||
@@ -254,8 +254,6 @@ github.com/pion/turn/v2 v2.0.5 h1:iwMHqDfPEDEOFzwWKT56eFmh6DYC6o/+xnLAEzgISbA=
|
||||
github.com/pion/turn/v2 v2.0.5/go.mod h1:APg43CFyt/14Uy7heYUOGWdkem/Wu4PhCO/bjyrTqMw=
|
||||
github.com/pion/udp v0.1.0 h1:uGxQsNyrqG3GLINv36Ff60covYmfrLoxzwnCsIYspXI=
|
||||
github.com/pion/udp v0.1.0/go.mod h1:BPELIjbwE9PRbd/zxI/KYBnbo7B6+oA6YuEaNE8lths=
|
||||
github.com/pion/webrtc/v3 v3.0.0-beta.7 h1:wDzex7YkT3wErRGaW4yOs9ziCQeuVF04poWVTqQ8o/Q=
|
||||
github.com/pion/webrtc/v3 v3.0.0-beta.7/go.mod h1:g2FHuaaKKuoZ9RlKIDAYvnO/rhocYG/d+2Ly1hRa85o=
|
||||
github.com/pion/webrtc/v3 v3.0.0-beta.9 h1:Vwep6pua7wJ5lSnR78KlSu0N3u9ZCBGfD65r5FwO35k=
|
||||
github.com/pion/webrtc/v3 v3.0.0-beta.9/go.mod h1:GlriYYHJ5KkNsCunm3oFDPql4TDTrrNoI9iSWWSnafA=
|
||||
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
@@ -470,6 +468,7 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
|
||||
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
|
||||
golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
|
||||
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
@@ -498,6 +497,7 @@ golang.org/x/tools v0.0.0-20191012152004-8de300cfc20a/go.mod h1:b+2E5dAYhXwXZwtn
|
||||
golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5 h1:hKsoRgsbwY1NafxrwTs+k64bikrLBkAgPir1TNCj3Zs=
|
||||
golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc h1:NCy3Ohtk6Iny5V/reW2Ktypo4zIpWBdRJ1uFMjBxdg8=
|
||||
golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
|
||||
+8
-2
@@ -15,6 +15,10 @@ type WebRTCConfig struct {
|
||||
setting webrtc.SettingEngine
|
||||
feedbackTypes []webrtc.RTCPFeedback
|
||||
|
||||
receiver ReceiverConfig
|
||||
}
|
||||
|
||||
type ReceiverConfig struct {
|
||||
maxBandwidth uint64
|
||||
maxBufferTime int
|
||||
}
|
||||
@@ -67,7 +71,9 @@ func NewWebRTCConfig(conf *config.RTCConfig, externalIP string) (*WebRTCConfig,
|
||||
configuration: c,
|
||||
setting: s,
|
||||
feedbackTypes: ft,
|
||||
maxBandwidth: conf.MaxBandwidth,
|
||||
maxBufferTime: conf.MaxBufferTime,
|
||||
receiver: ReceiverConfig{
|
||||
maxBandwidth: conf.MaxBandwidth,
|
||||
maxBufferTime: conf.MaxBufferTime,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
+12
-4
@@ -18,6 +18,7 @@ const (
|
||||
type MediaEngine struct {
|
||||
webrtc.MediaEngine
|
||||
feedbackTypes []webrtc.RTCPFeedback
|
||||
tCCExt int
|
||||
}
|
||||
|
||||
// PopulateFromSDP finds all codecs in sd and adds them to m, using the dynamic
|
||||
@@ -27,16 +28,23 @@ type MediaEngine struct {
|
||||
// PopulateFromSDP allows an answerer to properly match the PayloadTypes from the offerer.
|
||||
// A MediaEngine populated by PopulateFromSDP should be used only for a single session.
|
||||
func (e *MediaEngine) PopulateFromSDP(sd webrtc.SessionDescription) error {
|
||||
sdp := sdp.SessionDescription{}
|
||||
if err := sdp.Unmarshal([]byte(sd.SDP)); err != nil {
|
||||
s := sdp.SessionDescription{}
|
||||
if err := s.Unmarshal([]byte(sd.SDP)); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, md := range sdp.MediaDescriptions {
|
||||
for _, md := range s.MediaDescriptions {
|
||||
if md.MediaName.Media != mediaNameAudio && md.MediaName.Media != mediaNameVideo {
|
||||
continue
|
||||
}
|
||||
|
||||
for _, att := range md.Attributes {
|
||||
if att.Key == sdp.AttrKeyExtMap && strings.HasSuffix(att.Value, sdp.TransportCCURI) {
|
||||
e.tCCExt, _ = strconv.Atoi(att.Value[:1])
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
for _, format := range md.MediaName.Formats {
|
||||
pt, err := strconv.Atoi(format)
|
||||
if err != nil {
|
||||
@@ -44,7 +52,7 @@ func (e *MediaEngine) PopulateFromSDP(sd webrtc.SessionDescription) error {
|
||||
}
|
||||
|
||||
payloadType := uint8(pt)
|
||||
payloadCodec, err := sdp.GetCodecForPayloadType(payloadType)
|
||||
payloadCodec, err := s.GetCodecForPayloadType(payloadType)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not find codec for payload type %d", payloadType)
|
||||
}
|
||||
|
||||
+12
-9
@@ -7,19 +7,19 @@ import (
|
||||
)
|
||||
|
||||
type WebRTCPeer struct {
|
||||
id string
|
||||
conn *webrtc.PeerConnection
|
||||
ctx context.Context
|
||||
cancelFunc context.CancelFunc
|
||||
mediaEngine MediaEngine
|
||||
|
||||
maxBandwidth uint64
|
||||
maxBufferTime int
|
||||
receiverConfig ReceiverConfig
|
||||
|
||||
// callbacks & handlers
|
||||
onPeerTrack func(*PeerTrack)
|
||||
}
|
||||
|
||||
func NewWebRTCPeer(me MediaEngine, conf WebRTCConfig) (*WebRTCPeer, error) {
|
||||
func NewWebRTCPeer(id string, me MediaEngine, conf WebRTCConfig) (*WebRTCPeer, error) {
|
||||
api := webrtc.NewAPI(webrtc.WithMediaEngine(me.MediaEngine), webrtc.WithSettingEngine(conf.setting))
|
||||
pc, err := api.NewPeerConnection(conf.configuration)
|
||||
|
||||
@@ -29,12 +29,12 @@ func NewWebRTCPeer(me MediaEngine, conf WebRTCConfig) (*WebRTCPeer, error) {
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
peer := &WebRTCPeer{
|
||||
conn: pc,
|
||||
ctx: ctx,
|
||||
cancelFunc: cancel,
|
||||
mediaEngine: me,
|
||||
maxBandwidth: conf.maxBandwidth,
|
||||
maxBufferTime: conf.maxBufferTime,
|
||||
id: id,
|
||||
conn: pc,
|
||||
ctx: ctx,
|
||||
cancelFunc: cancel,
|
||||
mediaEngine: me,
|
||||
receiverConfig: conf.receiver,
|
||||
}
|
||||
|
||||
pc.OnTrack(peer.onTrack)
|
||||
@@ -44,5 +44,8 @@ func NewWebRTCPeer(me MediaEngine, conf WebRTCConfig) (*WebRTCPeer, error) {
|
||||
|
||||
// when a new track is created, creates a PeerTrack and adds it to room
|
||||
func (p *WebRTCPeer) onTrack(track *webrtc.Track, receiver *webrtc.RTPReceiver) {
|
||||
onPeerTrack := p.onPeerTrack
|
||||
if onPeerTrack != nil {
|
||||
|
||||
}
|
||||
}
|
||||
+118
-2
@@ -1,13 +1,129 @@
|
||||
package rtc
|
||||
|
||||
import (
|
||||
sfu "github.com/pion/ion-sfu/pkg"
|
||||
"io"
|
||||
|
||||
"github.com/pion/rtp"
|
||||
|
||||
"github.com/livekit/livekit-server/pkg/sfu"
|
||||
|
||||
"context"
|
||||
|
||||
"github.com/pion/rtcp"
|
||||
"github.com/pion/webrtc/v3"
|
||||
)
|
||||
|
||||
const (
|
||||
// TODO: could probably increase this depending on configuration/memory
|
||||
maxChanSize = 1024
|
||||
)
|
||||
|
||||
// A receiver is responsible for pulling from a track
|
||||
type Receiver struct {
|
||||
track *webrtc.Track
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
rtpReceiver *webrtc.RTPReceiver
|
||||
track *webrtc.Track
|
||||
buffer *sfu.Buffer
|
||||
rtpChan chan *rtp.Packet
|
||||
|
||||
onCloseHandler func(r *Receiver)
|
||||
}
|
||||
|
||||
func NewReceiver(ctx context.Context, rtpReceiver *webrtc.RTPReceiver, conf ReceiverConfig, me *MediaEngine) *Receiver {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
track := rtpReceiver.Track()
|
||||
return &Receiver{
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
rtpReceiver: rtpReceiver,
|
||||
track: track,
|
||||
rtpChan: make(chan *rtp.Packet, maxChanSize),
|
||||
buffer: sfu.NewBuffer(track, sfu.BufferOptions{
|
||||
BufferTime: conf.maxBufferTime,
|
||||
MaxBitRate: conf.maxBandwidth * 1000,
|
||||
TCCExt: me.tCCExt,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
// starts reading RTP and push to buffer
|
||||
func (r *Receiver) Start() {
|
||||
|
||||
}
|
||||
|
||||
// close the stream
|
||||
func (r *Receiver) Close() {
|
||||
|
||||
}
|
||||
|
||||
// reaturns channel to read rtp packets
|
||||
func (r *Receiver) RTPChan() <-chan *rtp.Packet {
|
||||
return r.rtpChan
|
||||
}
|
||||
|
||||
// Builds RTCP report from buffer
|
||||
func (r *Receiver) BuildRTCP() (rtcp.ReceptionReport, []rtcp.Packet) {
|
||||
return r.buffer.BuildRTCP()
|
||||
}
|
||||
|
||||
// WriteBufferedPacket writes buffered packet to track, return error if packet not found
|
||||
func (r *Receiver) WriteBufferedPacket(sn uint16, track *webrtc.Track, snOffset uint16, tsOffset, ssrc uint32) error {
|
||||
if r.buffer == nil || r.ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
return r.buffer.WritePacket(sn, track, snOffset, tsOffset, ssrc)
|
||||
}
|
||||
|
||||
// rtpWorker reads RTP stream, fills buffer and channel
|
||||
func (r *Receiver) rtpWorker() {
|
||||
defer func() {
|
||||
close(r.rtpChan)
|
||||
if r.onCloseHandler != nil {
|
||||
r.onCloseHandler(r)
|
||||
}
|
||||
}()
|
||||
|
||||
for {
|
||||
pkt, err := r.track.ReadRTP()
|
||||
if err == io.EOF {
|
||||
r.Close()
|
||||
return
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
// log and continue
|
||||
continue
|
||||
}
|
||||
|
||||
r.buffer.Push(pkt)
|
||||
|
||||
select {
|
||||
case <-r.ctx.Done():
|
||||
return
|
||||
default:
|
||||
r.rtpChan <- pkt
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// rtcpWorker reads RTCP messages from receiver, notifies buffer
|
||||
func (r *Receiver) rtcpWorker() {
|
||||
for {
|
||||
pkts, err := r.rtpReceiver.ReadRTCP()
|
||||
if err == io.ErrClosedPipe || r.ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
// TODO: log
|
||||
continue
|
||||
}
|
||||
|
||||
for _, pkt := range pkts {
|
||||
switch pkt := pkt.(type) {
|
||||
case *rtcp.SenderReport:
|
||||
r.buffer.SetSenderReportData(pkt.RTPTime, pkt.NTPTime)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+13
-3
@@ -3,7 +3,6 @@ package rtc
|
||||
import (
|
||||
"sync"
|
||||
|
||||
sfu "github.com/pion/ion-sfu/pkg"
|
||||
"github.com/pion/webrtc/v3"
|
||||
)
|
||||
|
||||
@@ -11,17 +10,28 @@ import (
|
||||
* Peer track represents a track that needs to be forwarded
|
||||
*/
|
||||
type PeerTrack struct {
|
||||
id string
|
||||
id uint32
|
||||
peerId string
|
||||
// source track
|
||||
track *webrtc.Track
|
||||
lock sync.RWMutex
|
||||
// map of target peerId -> forwarder
|
||||
forwarders map[string]Forwarder
|
||||
receiver sfu.Receiver
|
||||
receiver Receiver
|
||||
lastNack int64
|
||||
}
|
||||
|
||||
func NewPeerTrack(peerId string, track *webrtc.Track, receiver Receiver) *PeerTrack {
|
||||
return &PeerTrack{
|
||||
id: track.SSRC(),
|
||||
peerId: peerId,
|
||||
track: track,
|
||||
lock: sync.RWMutex{},
|
||||
forwarders: make(map[string]Forwarder),
|
||||
receiver: receiver,
|
||||
}
|
||||
}
|
||||
|
||||
// subscribes peer to current track
|
||||
// creates and add necessary forwarders and starts them
|
||||
func (t *PeerTrack) AddSubscriber(peer *WebRTCPeer) error {
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
This package is largely files from the wonderful ion-sfu project.
|
||||
|
||||
https://github.com/pion/ion-sfu
|
||||
|
||||
It's duplicated here since we needed to access a private method in the Buffer class
|
||||
@@ -0,0 +1,442 @@
|
||||
package sfu
|
||||
|
||||
// file from ion-sfu project
|
||||
|
||||
import (
|
||||
"math"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gammazero/deque"
|
||||
"github.com/pion/rtcp"
|
||||
"github.com/pion/rtp"
|
||||
"github.com/pion/webrtc/v3"
|
||||
)
|
||||
|
||||
const (
|
||||
maxSN = 1 << 16
|
||||
// default buffer time by ms
|
||||
defaultBufferTime = 1000
|
||||
)
|
||||
|
||||
type rtpExtInfo struct {
|
||||
ExtTSN uint32
|
||||
Timestamp int64
|
||||
}
|
||||
|
||||
// Buffer contains all packets
|
||||
type Buffer struct {
|
||||
mu sync.RWMutex
|
||||
|
||||
pktQueue queue
|
||||
codecType webrtc.RTPCodecType
|
||||
simulcast bool
|
||||
clockRate uint32
|
||||
maxBitrate uint64
|
||||
|
||||
// supported feedbacks
|
||||
remb bool
|
||||
nack bool
|
||||
tcc bool
|
||||
|
||||
lastSRNTPTime uint64
|
||||
lastSRRTPTime uint32
|
||||
lastSRRecv int64 // Represents wall clock of the most recent sender report arrival
|
||||
baseSN uint16
|
||||
cycles uint32
|
||||
lastExpected uint32
|
||||
lastReceived uint32
|
||||
lostRate float32
|
||||
ssrc uint32
|
||||
lastPacketTime int64 // Time the last RTP packet from this source was received
|
||||
lastRtcpPacketTime int64 // Time the last RTCP packet was received.
|
||||
lastRtcpSrTime int64 // Time the last RTCP SR was received. Required for DLSR computation.
|
||||
packetCount uint32 // Number of packets received from this source.
|
||||
lastTransit uint32
|
||||
maxSeqNo uint16 // The highest sequence number received in an RTP data packet
|
||||
jitter float64 // An estimate of the statistical variance of the RTP data packet inter-arrival time.
|
||||
totalByte uint64
|
||||
|
||||
// remb
|
||||
rembSteps uint8
|
||||
|
||||
// transport-cc
|
||||
tccExt uint8
|
||||
tccExtInfo []rtpExtInfo
|
||||
tccCycles uint32
|
||||
tccLastExtSN uint32
|
||||
tccPktCtn uint8
|
||||
tccLastSn uint16
|
||||
lastExtInfo uint16
|
||||
}
|
||||
|
||||
// BufferOptions provides configuration options for the buffer
|
||||
type BufferOptions struct {
|
||||
TCCExt int
|
||||
BufferTime int
|
||||
MaxBitRate uint64
|
||||
}
|
||||
|
||||
// NewBuffer constructs a new Buffer
|
||||
func NewBuffer(track *webrtc.Track, o BufferOptions) *Buffer {
|
||||
b := &Buffer{
|
||||
ssrc: track.SSRC(),
|
||||
clockRate: track.Codec().ClockRate,
|
||||
codecType: track.Codec().Type,
|
||||
maxBitrate: o.MaxBitRate,
|
||||
simulcast: len(track.RID()) > 0,
|
||||
rembSteps: 4,
|
||||
}
|
||||
if o.BufferTime <= 0 {
|
||||
o.BufferTime = defaultBufferTime
|
||||
}
|
||||
b.pktQueue.duration = uint32(o.BufferTime) * b.clockRate / 1000
|
||||
b.pktQueue.ssrc = track.SSRC()
|
||||
b.tccExt = uint8(o.TCCExt)
|
||||
|
||||
for _, fb := range track.Codec().RTCPFeedback {
|
||||
switch fb.Type {
|
||||
case webrtc.TypeRTCPFBGoogREMB:
|
||||
//log.Debugf("Setting feedback %s", webrtc.TypeRTCPFBGoogREMB)
|
||||
b.remb = true
|
||||
case webrtc.TypeRTCPFBTransportCC:
|
||||
//log.Debugf("Setting feedback %s", webrtc.TypeRTCPFBTransportCC)
|
||||
b.tccExtInfo = make([]rtpExtInfo, 1<<8)
|
||||
b.tcc = true
|
||||
case webrtc.TypeRTCPFBNACK:
|
||||
//log.Debugf("Setting feedback %s", webrtc.TypeRTCPFBNACK)
|
||||
b.nack = true
|
||||
}
|
||||
}
|
||||
//log.Debugf("NewBuffer BufferOptions=%v", o)
|
||||
return b
|
||||
}
|
||||
|
||||
// Push adds a RTP Packet, out of order, new packet may be arrived later
|
||||
func (b *Buffer) Push(p *rtp.Packet) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.totalByte += uint64(p.MarshalSize())
|
||||
if b.packetCount == 0 {
|
||||
b.baseSN = p.SequenceNumber
|
||||
b.maxSeqNo = p.SequenceNumber
|
||||
b.pktQueue.headSN = p.SequenceNumber - 1
|
||||
} else if snDiff(b.maxSeqNo, p.SequenceNumber) <= 0 {
|
||||
if p.SequenceNumber < b.maxSeqNo {
|
||||
b.cycles += maxSN
|
||||
}
|
||||
b.maxSeqNo = p.SequenceNumber
|
||||
}
|
||||
b.packetCount++
|
||||
b.lastPacketTime = time.Now().UnixNano()
|
||||
arrival := uint32(b.lastPacketTime / 1e6 * int64(b.clockRate/1e3))
|
||||
transit := arrival - p.Timestamp
|
||||
if b.lastTransit != 0 {
|
||||
d := int32(transit - b.lastTransit)
|
||||
if d < 0 {
|
||||
d = -d
|
||||
}
|
||||
b.jitter += (float64(d) - b.jitter) / 16
|
||||
}
|
||||
b.lastTransit = transit
|
||||
if b.codecType == webrtc.RTPCodecTypeVideo {
|
||||
b.pktQueue.AddPacket(p, p.SequenceNumber == b.maxSeqNo)
|
||||
}
|
||||
|
||||
if b.tcc {
|
||||
rtpTCC := rtp.TransportCCExtension{}
|
||||
if err := rtpTCC.Unmarshal(p.GetExtension(b.tccExt)); err == nil {
|
||||
if rtpTCC.TransportSequence < 0x0fff && (b.tccLastSn&0xffff) > 0xf000 {
|
||||
b.tccCycles += maxSN
|
||||
}
|
||||
b.tccExtInfo = append(b.tccExtInfo, rtpExtInfo{
|
||||
ExtTSN: b.tccCycles | uint32(rtpTCC.TransportSequence),
|
||||
Timestamp: b.lastPacketTime / 1e3,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (b *Buffer) buildREMBPacket() *rtcp.ReceiverEstimatedMaximumBitrate {
|
||||
br := b.maxBitrate
|
||||
if b.rembSteps > 0 {
|
||||
br /= uint64(b.rembSteps)
|
||||
b.rembSteps--
|
||||
}
|
||||
|
||||
return &rtcp.ReceiverEstimatedMaximumBitrate{
|
||||
SenderSSRC: b.ssrc,
|
||||
Bitrate: br,
|
||||
SSRCs: []uint32{b.ssrc},
|
||||
}
|
||||
}
|
||||
|
||||
func (b *Buffer) buildTransportCCPacket() *rtcp.TransportLayerCC {
|
||||
if len(b.tccExtInfo) == 0 {
|
||||
return nil
|
||||
}
|
||||
sort.Slice(b.tccExtInfo, func(i, j int) bool {
|
||||
return b.tccExtInfo[i].ExtTSN < b.tccExtInfo[j].ExtTSN
|
||||
})
|
||||
tccPkts := make([]rtpExtInfo, 0, int(float64(len(b.tccExtInfo))*1.2))
|
||||
for _, tccExtInfo := range b.tccExtInfo {
|
||||
if tccExtInfo.ExtTSN < b.tccLastExtSN {
|
||||
continue
|
||||
}
|
||||
if b.tccLastExtSN != 0 {
|
||||
for j := b.tccLastExtSN + 1; j < tccExtInfo.ExtTSN; j++ {
|
||||
tccPkts = append(tccPkts, rtpExtInfo{ExtTSN: j})
|
||||
}
|
||||
}
|
||||
b.tccLastExtSN = tccExtInfo.ExtTSN
|
||||
tccPkts = append(tccPkts, tccExtInfo)
|
||||
}
|
||||
b.tccExtInfo = b.tccExtInfo[:0]
|
||||
|
||||
rtcpTCC := &rtcp.TransportLayerCC{
|
||||
Header: rtcp.Header{
|
||||
Padding: true,
|
||||
Count: rtcp.FormatTCC,
|
||||
Type: rtcp.TypeTransportSpecificFeedback,
|
||||
},
|
||||
MediaSSRC: b.ssrc,
|
||||
BaseSequenceNumber: uint16(tccPkts[0].ExtTSN),
|
||||
PacketStatusCount: uint16(len(tccPkts)),
|
||||
FbPktCount: b.tccPktCtn,
|
||||
}
|
||||
b.tccPktCtn++
|
||||
|
||||
firstRecv := false
|
||||
allSame := true
|
||||
timestamp := int64(0)
|
||||
deltaLen := 0
|
||||
lastStatus := rtcp.TypeTCCPacketReceivedWithoutDelta
|
||||
maxStatus := rtcp.TypeTCCPacketNotReceived
|
||||
|
||||
var statusList deque.Deque
|
||||
|
||||
for _, stat := range tccPkts {
|
||||
status := rtcp.TypeTCCPacketNotReceived
|
||||
if stat.Timestamp != 0 {
|
||||
var delta int64
|
||||
if !firstRecv {
|
||||
firstRecv = true
|
||||
timestamp = stat.Timestamp
|
||||
rtcpTCC.ReferenceTime = uint32(stat.Timestamp / 64000)
|
||||
}
|
||||
|
||||
delta = (stat.Timestamp - timestamp) / 250
|
||||
if delta < 0 || delta > 255 {
|
||||
status = rtcp.TypeTCCPacketReceivedLargeDelta
|
||||
rDelta := int16(delta)
|
||||
if int64(rDelta) != delta {
|
||||
if rDelta > 0 {
|
||||
rDelta = math.MaxInt16
|
||||
} else {
|
||||
rDelta = math.MinInt16
|
||||
}
|
||||
}
|
||||
rtcpTCC.RecvDeltas = append(rtcpTCC.RecvDeltas, &rtcp.RecvDelta{
|
||||
Type: status,
|
||||
Delta: int64(rDelta) * 250,
|
||||
})
|
||||
deltaLen += 2
|
||||
} else {
|
||||
status = rtcp.TypeTCCPacketReceivedSmallDelta
|
||||
rtcpTCC.RecvDeltas = append(rtcpTCC.RecvDeltas, &rtcp.RecvDelta{
|
||||
Type: status,
|
||||
Delta: delta * 250,
|
||||
})
|
||||
deltaLen++
|
||||
}
|
||||
timestamp = stat.Timestamp
|
||||
}
|
||||
|
||||
if allSame && lastStatus != rtcp.TypeTCCPacketReceivedWithoutDelta && status != lastStatus {
|
||||
if statusList.Len() > 7 {
|
||||
rtcpTCC.PacketChunks = append(rtcpTCC.PacketChunks, &rtcp.RunLengthChunk{
|
||||
PacketStatusSymbol: lastStatus,
|
||||
RunLength: uint16(statusList.Len()),
|
||||
})
|
||||
statusList.Clear()
|
||||
lastStatus = rtcp.TypeTCCPacketReceivedWithoutDelta
|
||||
maxStatus = rtcp.TypeTCCPacketNotReceived
|
||||
allSame = true
|
||||
} else {
|
||||
allSame = false
|
||||
}
|
||||
}
|
||||
statusList.PushBack(status)
|
||||
if status > maxStatus {
|
||||
maxStatus = status
|
||||
}
|
||||
lastStatus = status
|
||||
|
||||
if !allSame {
|
||||
if maxStatus == rtcp.TypeTCCPacketReceivedLargeDelta && statusList.Len() > 6 {
|
||||
symbolList := make([]uint16, 7)
|
||||
for i := 0; i < 7; i++ {
|
||||
symbolList[i] = statusList.PopFront().(uint16)
|
||||
}
|
||||
rtcpTCC.PacketChunks = append(rtcpTCC.PacketChunks, &rtcp.StatusVectorChunk{
|
||||
SymbolSize: rtcp.TypeTCCSymbolSizeTwoBit,
|
||||
SymbolList: symbolList,
|
||||
})
|
||||
lastStatus = rtcp.TypeTCCPacketReceivedWithoutDelta
|
||||
maxStatus = rtcp.TypeTCCPacketNotReceived
|
||||
allSame = true
|
||||
|
||||
for i := 0; i < statusList.Len(); i++ {
|
||||
status = statusList.At(i).(uint16)
|
||||
if status > maxStatus {
|
||||
maxStatus = status
|
||||
}
|
||||
if allSame && lastStatus != rtcp.TypeTCCPacketReceivedWithoutDelta && status != lastStatus {
|
||||
allSame = false
|
||||
}
|
||||
lastStatus = status
|
||||
}
|
||||
} else if statusList.Len() > 13 {
|
||||
symbolList := make([]uint16, 14)
|
||||
for i := 0; i < 14; i++ {
|
||||
symbolList[i] = statusList.PopFront().(uint16)
|
||||
}
|
||||
rtcpTCC.PacketChunks = append(rtcpTCC.PacketChunks, &rtcp.StatusVectorChunk{
|
||||
SymbolSize: rtcp.TypeTCCSymbolSizeOneBit,
|
||||
SymbolList: symbolList,
|
||||
})
|
||||
lastStatus = rtcp.TypeTCCPacketReceivedWithoutDelta
|
||||
maxStatus = rtcp.TypeTCCPacketNotReceived
|
||||
allSame = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if statusList.Len() > 0 {
|
||||
if allSame {
|
||||
rtcpTCC.PacketChunks = append(rtcpTCC.PacketChunks, &rtcp.RunLengthChunk{
|
||||
PacketStatusSymbol: lastStatus,
|
||||
RunLength: uint16(statusList.Len()),
|
||||
})
|
||||
} else if maxStatus == rtcp.TypeTCCPacketReceivedLargeDelta {
|
||||
symbolList := make([]uint16, statusList.Len())
|
||||
for i := 0; i < statusList.Len(); i++ {
|
||||
symbolList[i] = statusList.PopFront().(uint16)
|
||||
}
|
||||
rtcpTCC.PacketChunks = append(rtcpTCC.PacketChunks, &rtcp.StatusVectorChunk{
|
||||
SymbolSize: rtcp.TypeTCCSymbolSizeTwoBit,
|
||||
SymbolList: symbolList,
|
||||
})
|
||||
} else {
|
||||
symbolList := make([]uint16, statusList.Len())
|
||||
for i := 0; i < statusList.Len(); i++ {
|
||||
symbolList[i] = statusList.PopFront().(uint16)
|
||||
}
|
||||
rtcpTCC.PacketChunks = append(rtcpTCC.PacketChunks, &rtcp.StatusVectorChunk{
|
||||
SymbolSize: rtcp.TypeTCCSymbolSizeOneBit,
|
||||
SymbolList: symbolList,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pLen := uint16(20 + len(rtcpTCC.PacketChunks)*2 + deltaLen)
|
||||
rtcpTCC.Header.Padding = pLen%4 != 0
|
||||
for pLen%4 != 0 {
|
||||
pLen++
|
||||
}
|
||||
rtcpTCC.Header.Length = (pLen / 4) - 1
|
||||
return rtcpTCC
|
||||
}
|
||||
|
||||
func (b *Buffer) buildReceptionReport() rtcp.ReceptionReport {
|
||||
extMaxSeq := b.cycles | uint32(b.maxSeqNo)
|
||||
expected := extMaxSeq - uint32(b.baseSN) + 1
|
||||
lost := expected - b.packetCount
|
||||
if b.packetCount == 0 {
|
||||
lost = 0
|
||||
}
|
||||
expectedInterval := expected - b.lastExpected
|
||||
b.lastExpected = expected
|
||||
|
||||
receivedInterval := b.packetCount - b.lastReceived
|
||||
b.lastReceived = b.packetCount
|
||||
|
||||
lostInterval := expectedInterval - receivedInterval
|
||||
|
||||
b.lostRate = float32(lostInterval) / float32(expectedInterval)
|
||||
var fracLost uint8
|
||||
if expectedInterval != 0 && lostInterval > 0 {
|
||||
fracLost = uint8((lostInterval << 8) / expectedInterval)
|
||||
}
|
||||
var dlsr uint32
|
||||
if b.lastSRRecv != 0 {
|
||||
delayMS := uint32((time.Now().UnixNano() - b.lastSRRecv) / 1e6)
|
||||
dlsr = (delayMS / 1e3) << 16
|
||||
dlsr |= (delayMS % 1e3) * 65536 / 1000
|
||||
}
|
||||
|
||||
rr := rtcp.ReceptionReport{
|
||||
SSRC: b.ssrc,
|
||||
FractionLost: fracLost,
|
||||
TotalLost: lost,
|
||||
LastSequenceNumber: extMaxSeq,
|
||||
Jitter: uint32(b.jitter),
|
||||
LastSenderReport: uint32(b.lastSRNTPTime >> 16),
|
||||
Delay: dlsr,
|
||||
}
|
||||
return rr
|
||||
}
|
||||
|
||||
func (b *Buffer) SetSenderReportData(rtpTime uint32, ntpTime uint64) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.lastSRRTPTime = rtpTime
|
||||
b.lastSRNTPTime = ntpTime
|
||||
b.lastSRRecv = time.Now().UnixNano()
|
||||
}
|
||||
|
||||
func (b *Buffer) BuildRTCP() (rtcp.ReceptionReport, []rtcp.Packet) {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
var pkts []rtcp.Packet
|
||||
var report rtcp.ReceptionReport
|
||||
|
||||
report = b.buildReceptionReport()
|
||||
|
||||
if b.remb {
|
||||
pkts = append(pkts, b.buildREMBPacket())
|
||||
}
|
||||
|
||||
if b.tcc {
|
||||
if tccPkt := b.buildTransportCCPacket(); tccPkt != nil {
|
||||
pkts = append(pkts, tccPkt)
|
||||
}
|
||||
}
|
||||
|
||||
return report, pkts
|
||||
}
|
||||
|
||||
// WritePacket write buffer packet to requested track. and modify headers
|
||||
func (b *Buffer) WritePacket(sn uint16, track *webrtc.Track, snOffset uint16, tsOffset, ssrc uint32) error {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
if bufferPkt := b.pktQueue.GetPacket(sn); bufferPkt != nil {
|
||||
bSsrc := bufferPkt.SSRC
|
||||
bufferPkt.SequenceNumber -= snOffset
|
||||
bufferPkt.Timestamp -= tsOffset
|
||||
bufferPkt.SSRC = ssrc
|
||||
err := track.WriteRTP(bufferPkt)
|
||||
bufferPkt.Timestamp += tsOffset
|
||||
bufferPkt.SequenceNumber += snOffset
|
||||
bufferPkt.SSRC = bSsrc
|
||||
return err
|
||||
}
|
||||
return errPacketNotFound
|
||||
}
|
||||
|
||||
func (b *Buffer) onLostHandler(fn func(nack *rtcp.TransportLayerNack)) {
|
||||
if b.nack {
|
||||
b.pktQueue.onLost = fn
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package sfu
|
||||
|
||||
import "errors"
|
||||
|
||||
var (
|
||||
errPeerConnectionInitFailed = errors.New("pc init failed")
|
||||
errPtNotSupported = errors.New("payload type not supported")
|
||||
// router errors
|
||||
errNoReceiverFound = errors.New("no receiver found")
|
||||
// Helpers errors
|
||||
errShortPacket = errors.New("packet is not large enough")
|
||||
errNilPacket = errors.New("invalid nil packet")
|
||||
// buffer errors
|
||||
errPacketNotFound = errors.New("packet not found in cache")
|
||||
errPacketTooOld = errors.New("packet not found in cache, too old")
|
||||
)
|
||||
@@ -0,0 +1,134 @@
|
||||
package sfu
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
const ntpEpoch = 2208988800
|
||||
|
||||
type atomicBool struct {
|
||||
val int32
|
||||
}
|
||||
|
||||
func (b *atomicBool) set(value bool) { // nolint: unparam
|
||||
var i int32
|
||||
if value {
|
||||
i = 1
|
||||
}
|
||||
|
||||
atomic.StoreInt32(&(b.val), i)
|
||||
}
|
||||
|
||||
func (b *atomicBool) get() bool {
|
||||
return atomic.LoadInt32(&(b.val)) != 0
|
||||
}
|
||||
|
||||
// VP8Helper is a helper to get temporal data from VP8 packet header
|
||||
/*
|
||||
VP8Helper Payload Descriptor
|
||||
0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7
|
||||
+-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+
|
||||
|X|R|N|S|R| PID | (REQUIRED) |X|R|N|S|R| PID | (REQUIRED)
|
||||
+-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+
|
||||
X: |I|L|T|K| RSV | (OPTIONAL) X: |I|L|T|K| RSV | (OPTIONAL)
|
||||
+-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+
|
||||
I: |M| PictureID | (OPTIONAL) I: |M| PictureID | (OPTIONAL)
|
||||
+-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+
|
||||
L: | TL0PICIDX | (OPTIONAL) | PictureID |
|
||||
+-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+
|
||||
T/K:|TID|Y| KEYIDX | (OPTIONAL) L: | TL0PICIDX | (OPTIONAL)
|
||||
+-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+
|
||||
T/K:|TID|Y| KEYIDX | (OPTIONAL)
|
||||
+-+-+-+-+-+-+-+-+
|
||||
*/
|
||||
type VP8Helper struct {
|
||||
TemporalSupported bool
|
||||
// Optional Header
|
||||
PictureID uint16 /* 8 or 16 bits, picture ID */
|
||||
picIDIdx uint8
|
||||
mBit bool
|
||||
TL0PICIDX uint8 /* 8 bits temporal level zero index */
|
||||
tlzIdx uint8
|
||||
|
||||
// Optional Header If either of the T or K bits are set to 1,
|
||||
// the TID/Y/KEYIDX extension field MUST be present.
|
||||
TID uint8 /* 2 bits temporal layer idx*/
|
||||
// IsKeyFrame is a helper to detect if current packet is a keyframe
|
||||
IsKeyFrame bool
|
||||
}
|
||||
|
||||
// Unmarshal parses the passed byte slice and stores the result in the VP8Helper this method is called upon
|
||||
func (p *VP8Helper) Unmarshal(payload []byte) error {
|
||||
if payload == nil {
|
||||
return errNilPacket
|
||||
}
|
||||
|
||||
payloadLen := len(payload)
|
||||
|
||||
if payloadLen < 4 {
|
||||
return errShortPacket
|
||||
}
|
||||
|
||||
var idx uint8
|
||||
// Check for extended bit control
|
||||
if payload[idx]&0x80 > 0 {
|
||||
idx++
|
||||
// Check if T is present, if not, no temporal layer is available
|
||||
p.TemporalSupported = payload[idx]&0x20 > 0
|
||||
L := payload[idx]&0x40 > 0
|
||||
// Check for PictureID
|
||||
if payload[idx]&0x80 > 0 {
|
||||
idx++
|
||||
p.picIDIdx = idx
|
||||
pid := payload[idx] & 0x7f
|
||||
// Check if m is 1, then Picture ID is 15 bits
|
||||
if payload[idx]&0x80 > 0 {
|
||||
idx++
|
||||
p.mBit = true
|
||||
p.PictureID = binary.BigEndian.Uint16([]byte{pid, payload[idx]})
|
||||
} else {
|
||||
p.PictureID = uint16(pid)
|
||||
}
|
||||
}
|
||||
// Check if TL0PICIDX is present
|
||||
if L {
|
||||
idx++
|
||||
p.tlzIdx = idx
|
||||
p.TL0PICIDX = payload[idx]
|
||||
}
|
||||
idx++
|
||||
// Set TID
|
||||
p.TID = (payload[idx] & 0xc0) >> 6
|
||||
idx++
|
||||
if int(idx) >= payloadLen {
|
||||
return errShortPacket
|
||||
}
|
||||
// Check is packet is a keyframe by looking at P bit in vp8 payload
|
||||
p.IsKeyFrame = payload[idx]&0x01 == 0
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func snDiff(sn1, sn2 uint16) int {
|
||||
if sn1 == sn2 {
|
||||
return 0
|
||||
}
|
||||
if ((sn2 - sn1) & 0x8000) != 0 {
|
||||
return 1
|
||||
}
|
||||
return -1
|
||||
}
|
||||
|
||||
func timeToNtp(ns int64) uint64 {
|
||||
seconds := uint64(ns/1e9 + ntpEpoch)
|
||||
fraction := uint64(((ns % 1e9) << 32) / 1e9)
|
||||
return seconds<<32 | fraction
|
||||
}
|
||||
|
||||
// fromNtp converts a NTP timestamp into GO time
|
||||
func fromNtp(seconds, fraction uint32) (tm int64) {
|
||||
n := (int64(fraction) * 1e9) >> 32
|
||||
tm = (int64(seconds)-ntpEpoch)*1e9 + n
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
package sfu
|
||||
|
||||
import (
|
||||
//"github.com/pion/ion-sfu/pkg/log"
|
||||
"github.com/pion/rtcp"
|
||||
"github.com/pion/rtp"
|
||||
)
|
||||
|
||||
type queue struct {
|
||||
pkts []*rtp.Packet
|
||||
ssrc uint32
|
||||
head int
|
||||
tail int
|
||||
size int
|
||||
headSN uint16
|
||||
counter int
|
||||
duration uint32
|
||||
onLost func(nack *rtcp.TransportLayerNack)
|
||||
}
|
||||
|
||||
func (q *queue) AddPacket(pkt *rtp.Packet, latest bool) {
|
||||
if !latest {
|
||||
q.set(int(q.headSN-pkt.SequenceNumber), pkt)
|
||||
return
|
||||
}
|
||||
diff := pkt.SequenceNumber - q.headSN
|
||||
q.headSN = pkt.SequenceNumber
|
||||
for i := uint16(1); i < diff; i++ {
|
||||
q.push(nil)
|
||||
q.counter++
|
||||
}
|
||||
q.counter++
|
||||
q.push(pkt)
|
||||
if q.counter >= 7 {
|
||||
if n := q.nack(); n != nil && q.onLost != nil {
|
||||
q.onLost(&rtcp.TransportLayerNack{
|
||||
MediaSSRC: q.ssrc,
|
||||
Nacks: []rtcp.NackPair{*n},
|
||||
})
|
||||
}
|
||||
q.clean()
|
||||
q.counter -= 5
|
||||
}
|
||||
}
|
||||
|
||||
func (q *queue) GetPacket(sn uint16) *rtp.Packet {
|
||||
return q.get(int(q.headSN - sn))
|
||||
}
|
||||
|
||||
func (q *queue) push(pkt *rtp.Packet) {
|
||||
q.resize()
|
||||
q.head = (q.head - 1) & (len(q.pkts) - 1)
|
||||
q.pkts[q.head] = pkt
|
||||
q.size++
|
||||
}
|
||||
|
||||
func (q *queue) shift() {
|
||||
if q.size <= 0 {
|
||||
return
|
||||
}
|
||||
q.tail = (q.tail - 1) & (len(q.pkts) - 1)
|
||||
q.pkts[q.tail] = nil
|
||||
q.size--
|
||||
}
|
||||
|
||||
func (q *queue) last() *rtp.Packet {
|
||||
return q.pkts[(q.tail-1)&(len(q.pkts)-1)]
|
||||
}
|
||||
|
||||
func (q *queue) get(i int) *rtp.Packet {
|
||||
if i < 0 || i >= q.size {
|
||||
return nil
|
||||
}
|
||||
return q.pkts[(q.head+i)&(len(q.pkts)-1)]
|
||||
}
|
||||
|
||||
func (q *queue) set(i int, pkt *rtp.Packet) {
|
||||
if i < 0 || i >= q.size {
|
||||
//log.Warnf("warn: %v:", errPacketTooOld)
|
||||
return
|
||||
}
|
||||
q.pkts[(q.head+i)&(len(q.pkts)-1)] = pkt
|
||||
}
|
||||
|
||||
func (q *queue) resize() {
|
||||
if len(q.pkts) == 0 {
|
||||
q.pkts = make([]*rtp.Packet, 128)
|
||||
return
|
||||
}
|
||||
if q.size == len(q.pkts) {
|
||||
newBuf := make([]*rtp.Packet, q.size<<1)
|
||||
if q.tail > q.head {
|
||||
copy(newBuf, q.pkts[q.head:q.tail])
|
||||
} else {
|
||||
n := copy(newBuf, q.pkts[q.head:])
|
||||
copy(newBuf[n:], q.pkts[:q.tail])
|
||||
}
|
||||
q.head = 0
|
||||
q.tail = q.size
|
||||
q.pkts = newBuf
|
||||
}
|
||||
}
|
||||
|
||||
func (q *queue) nack() *rtcp.NackPair {
|
||||
for i := 0; i < 5; i++ {
|
||||
if q.get(q.counter-i-1) == nil {
|
||||
blp := uint16(0)
|
||||
for j := 1; j < q.counter-i; j++ {
|
||||
if q.get(q.counter-i-j-1) == nil {
|
||||
blp |= 1 << (j - 1)
|
||||
}
|
||||
}
|
||||
return &rtcp.NackPair{PacketID: q.headSN - uint16(q.counter-i-1), LostPackets: rtcp.PacketBitmap(blp)}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (q *queue) clean() {
|
||||
last := q.last()
|
||||
for q.size > 120 && (last == nil || q.pkts[q.head].Timestamp-last.Timestamp > q.duration) {
|
||||
q.shift()
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user