From 312a4f1def0e3949e36870d0011ebe65604c5d7e Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sun, 1 Nov 2020 20:44:33 -0800 Subject: [PATCH] filled in subscriber logic --- pkg/rtc/errors.go | 5 +-- pkg/rtc/forwarder.go | 52 ++++++++++++++++++++++------- pkg/rtc/peer.go | 2 +- pkg/rtc/track.go | 79 ++++++++++++++++++++++++++++++++++++++++---- 4 files changed, 116 insertions(+), 22 deletions(-) diff --git a/pkg/rtc/errors.go b/pkg/rtc/errors.go index 5b180bca7..6b92d8875 100644 --- a/pkg/rtc/errors.go +++ b/pkg/rtc/errors.go @@ -3,6 +3,7 @@ package rtc import "errors" var ( - ErrPeerExists = errors.New("peer already exists") - ErrPeerNotConnected = errors.New("peer has not been connected") + ErrPeerExists = errors.New("peer already exists") + ErrPeerNotConnected = errors.New("peer has not been connected") + ErrUnsupportedPayloadType = errors.New("peer does not support payload type") ) diff --git a/pkg/rtc/forwarder.go b/pkg/rtc/forwarder.go index ef359521c..de223be74 100644 --- a/pkg/rtc/forwarder.go +++ b/pkg/rtc/forwarder.go @@ -18,9 +18,14 @@ import ( type Forwarder interface { ID() string ChannelType() ChannelType + PeerID() string WriteRTP(*rtp.Packet) error + WriteRTCP(pkts []rtcp.Packet) error Start() Close() + CreatedAt() time.Time + + OnClose(func(Forwarder)) } type ChannelType int @@ -35,32 +40,41 @@ type SimpleForwarder struct { id string ctx context.Context cancel context.CancelFunc - peer *WebRTCPeer // source peer + peer *WebRTCPeer // destination peer sender *webrtc.RTPSender // destination sender track *webrtc.Track // sender track buffer *sfu.Buffer channelType ChannelType payload uint8 - lastPli time.Time + lastPli time.Time + createdAt time.Time once sync.Once + + // handlers + onClose func(forwarder Forwarder) } func NewSimpleForwarder(ctx context.Context, peer *WebRTCPeer, sender *webrtc.RTPSender, buffer *sfu.Buffer) *SimpleForwarder { ctx, cancel := context.WithCancel(ctx) f := &SimpleForwarder{ - id: peer.ID(), - ctx: ctx, - cancel: cancel, - peer: peer, - sender: sender, - buffer: buffer, - track: sender.Track(), - payload: sender.Track().PayloadType(), + id: peer.ID(), + ctx: ctx, + cancel: cancel, + peer: peer, + sender: sender, + buffer: buffer, + track: sender.Track(), + payload: sender.Track().PayloadType(), + createdAt: time.Now(), } - // TODO: set channel type + if f.track.Kind() == webrtc.RTPCodecTypeAudio { + f.channelType = ChannelAudio + } else if f.track.Kind() == webrtc.RTPCodecTypeVideo { + f.channelType = ChannelVideo + } return f } @@ -70,6 +84,10 @@ func (f *SimpleForwarder) ID() string { return f.id } +func (f *SimpleForwarder) PeerID() string { + return f.peer.ID() +} + func (f *SimpleForwarder) ChannelType() ChannelType { return f.channelType } @@ -85,7 +103,9 @@ func (f *SimpleForwarder) Close() { return } f.cancel() - // TODO: on close handler + if f.onClose != nil { + f.onClose(f) + } } // Writes an RTP packet to peer @@ -111,6 +131,14 @@ func (f *SimpleForwarder) WriteRTCP(pkts []rtcp.Packet) error { return f.peer.conn.WriteRTCP(pkts) } +func (f *SimpleForwarder) OnClose(closeFunc func(Forwarder)) { + f.onClose = closeFunc +} + +func (f *SimpleForwarder) CreatedAt() time.Time { + return f.createdAt +} + // rtcpWorker receives RTCP packets from the destination peer // this include packet loss packets func (f *SimpleForwarder) rtcpWorker() { diff --git a/pkg/rtc/peer.go b/pkg/rtc/peer.go index 31c5c346c..f260f9ebb 100644 --- a/pkg/rtc/peer.go +++ b/pkg/rtc/peer.go @@ -170,7 +170,7 @@ func (p *WebRTCPeer) onTrack(track *webrtc.Track, rtpReceiver *webrtc.RTPReceive // create Receiver receiver := NewReceiver(p.ctx, rtpReceiver, p.receiverConfig, p.mediaEngine) - pt := NewPeerTrack(p.id, track, receiver) + pt := NewPeerTrack(p.ctx, p.id, track, receiver) p.lock.Lock() p.tracks = append(p.tracks, pt) diff --git a/pkg/rtc/track.go b/pkg/rtc/track.go index c287c618f..817518405 100644 --- a/pkg/rtc/track.go +++ b/pkg/rtc/track.go @@ -1,16 +1,23 @@ package rtc import ( + "context" "sync" + "time" "github.com/pion/webrtc/v3" + + "github.com/livekit/livekit-server/pkg/logger" ) -/** - * Peer track represents a track that needs to be forwarded - */ +var ( + creationDelay = 500 * time.Millisecond +) + +// Peer track represents a track that needs to be forwarded type PeerTrack struct { id uint32 + ctx context.Context peerId string // source track track *webrtc.Track @@ -21,9 +28,10 @@ type PeerTrack struct { lastNack int64 } -func NewPeerTrack(peerId string, track *webrtc.Track, receiver *Receiver) *PeerTrack { +func NewPeerTrack(ctx context.Context, peerId string, track *webrtc.Track, receiver *Receiver) *PeerTrack { return &PeerTrack{ id: track.SSRC(), + ctx: ctx, peerId: peerId, track: track, lock: sync.RWMutex{}, @@ -35,6 +43,41 @@ func NewPeerTrack(peerId string, track *webrtc.Track, receiver *Receiver) *PeerT // subscribes peer to current track // creates and add necessary forwarders and starts them func (t *PeerTrack) AddSubscriber(peer *WebRTCPeer) error { + // check codecs supported by outbound peer + codecs := peer.mediaEngine.GetCodecsByName(t.track.Codec().Name) + if len(codecs) == 0 { + return ErrUnsupportedPayloadType + } + + // use existing SSRC with simple forwarders. adaptive forwarders require unique SSRC per layer + outTrack, err := peer.conn.NewTrack(codecs[0].PayloadType, t.track.SSRC(), t.track.ID(), t.track.Label()) + if err != nil { + return err + } + + rtpSender, err := peer.conn.AddTrack(outTrack) + if err != nil { + return err + } + + forwarder := NewSimpleForwarder(t.ctx, peer, rtpSender, t.receiver.buffer) + forwarder.OnClose(func(f Forwarder) { + t.lock.Lock() + delete(t.forwarders, peer.ID()) + t.lock.Unlock() + + if err := peer.conn.RemoveTrack(rtpSender); err != nil { + logger.GetLogger().Warnw("could not remove track from forwarder", + "peer", peer.ID()) + } + }) + + t.lock.Lock() + defer t.lock.Unlock() + t.forwarders[peer.ID()] = forwarder + + // start forwarder + forwarder.Start() return nil } @@ -42,12 +85,34 @@ func (t *PeerTrack) AddSubscriber(peer *WebRTCPeer) error { // removes peer from subscription // stop all forwarders to the peer func (t *PeerTrack) RemoveSubscriber(peerId string) { + t.lock.RLock() + defer t.lock.RUnlock() + if forwarder := t.forwarders[peerId]; forwarder != nil { + go forwarder.Close() + } } // forwardWorker reads from the receiver and writes to each sender func (t *PeerTrack) forwardWorker() { - t.lock.RLock() - defer t.lock.RUnlock() - + for pkt := range t.receiver.RTPChan() { + now := time.Now() + t.lock.RLock() + for _, forwarder := range t.forwarders { + // There exists a bug in chrome where setLocalDescription + // fails if track RTP arrives before the sfu offer is set. + // We delay sending RTP here to avoid the issue. + // https://bugs.chromium.org/p/webrtc/issues/detail?id=10139 + if now.Sub(forwarder.CreatedAt()) < creationDelay { + continue + } + if err := forwarder.WriteRTP(pkt); err != nil { + logger.GetLogger().Warnw("could not forward packet to peer", + "srcPeer", t.peerId, + "dstPeer", forwarder.PeerID(), + "track", forwarder.ID()) + } + } + t.lock.RUnlock() + } }