filled in subscriber logic

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")
 )

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() {

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)

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()
+	}
 }