send resync to ensure keyframes have been requested

go.mod

@@ -4,7 +4,6 @@ go 1.15
 
 require (
 	github.com/bep/debounce v1.2.0
-	github.com/gammazero/workerpool v1.1.1
 	github.com/go-redis/redis/v8 v8.4.8
 	github.com/golang/protobuf v1.4.3
 	github.com/google/wire v0.4.0
@@ -15,12 +14,11 @@ require (
 	github.com/magefile/mage v1.10.0
 	github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0
 	github.com/mitchellh/go-homedir v1.1.0
-	github.com/pion/ion-log v1.0.0
-	github.com/pion/ion-sfu v1.8.2
+	github.com/pion/ion-sfu v1.8.3
 	github.com/pion/rtcp v1.2.6
 	github.com/pion/rtp v1.6.2
+	github.com/pion/sdp/v3 v3.0.4
 	github.com/pion/stun v0.3.5
-	github.com/pion/transport v0.12.2
 	github.com/pion/webrtc/v3 v3.0.5
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.7.0
@@ -36,4 +34,4 @@ require (
 	gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776
 )
 
-replace github.com/pion/ion-sfu => github.com/davidzhao/ion-sfu v1.8.3-0.20210205014052-b5e378606cec
+replace github.com/pion/ion-sfu => ../ion-sfu

go.sum

@@ -71,10 +71,8 @@ github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7Do
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/davidzhao/ion-sfu v1.8.3-0.20210205010743-13f2b5e71196 h1:lYSupwHjpNUtXO3GEHjhX9S1F3efREDs4FpTAjf+Frc=
-github.com/davidzhao/ion-sfu v1.8.3-0.20210205010743-13f2b5e71196/go.mod h1:MKDh4JSLpBA/hFeYAYALb6nYpe6Ruknd3pmxPMbdpM8=
-github.com/davidzhao/ion-sfu v1.8.3-0.20210205014052-b5e378606cec h1:HjyyD0TXuCVheKu7HyWdye3jQ2/+dE/f2q7I+u3X2oA=
-github.com/davidzhao/ion-sfu v1.8.3-0.20210205014052-b5e378606cec/go.mod h1:MKDh4JSLpBA/hFeYAYALb6nYpe6Ruknd3pmxPMbdpM8=
+github.com/davidzhao/ion-sfu v1.8.3-0.20210205220645-c9ec7ad1db86 h1:POGjEQ8fyEQn96DMXMG7xhvWxHjNRMQ5aoiRzQOzOdY=
+github.com/davidzhao/ion-sfu v1.8.3-0.20210205220645-c9ec7ad1db86/go.mod h1:MKDh4JSLpBA/hFeYAYALb6nYpe6Ruknd3pmxPMbdpM8=
 github.com/desertbit/timer v0.0.0-20180107155436-c41aec40b27f/go.mod h1:xH/i4TFMt8koVQZ6WFms69WAsDWr2XsYL3Hkl7jkoLE=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
@@ -331,10 +329,6 @@ github.com/pion/interceptor v0.0.9 h1:fk5hTdyLO3KURQsf/+RjMpEm4NE3yeTY9Kh97b5Bvw
 github.com/pion/interceptor v0.0.9/go.mod h1:dHgEP5dtxOTf21MObuBAjJeAayPxLUAZjerGH8Xr07c=
 github.com/pion/ion-log v1.0.0 h1:2lJLImCmfCWCR38hLWsjQfBWe6NFz/htbqiYHwvOP/Q=
 github.com/pion/ion-log v1.0.0/go.mod h1:jwcla9KoB9bB/4FxYDSRJPcPYSLp5XiUUMnOLaqwl4E=
-github.com/pion/ion-sfu v1.8.1 h1:N3xaDZ2Om2pCzPLlusPmV7BJYVedE5VEx8um6d0Crns=
-github.com/pion/ion-sfu v1.8.1/go.mod h1:b6FI2WFTBsifw9EhxvfDks5/9SGYmVvm00IogZEgaik=
-github.com/pion/ion-sfu v1.8.2 h1:YIkelxhX5EZgWkmj6XY+0AVKi/se8W1bkOobEe2qjYc=
-github.com/pion/ion-sfu v1.8.2/go.mod h1:MKDh4JSLpBA/hFeYAYALb6nYpe6Ruknd3pmxPMbdpM8=
 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.4 h1:O4vvVqr4DGX63vzmO6Fw9vpy3lfztVWHGCQfyw0ZLSY=

pkg/config/config.go

@@ -25,7 +25,7 @@ type RTCConfig struct {
 	StunServers       []string `yaml:"stun_servers"`
 	UseExternalIP     bool     `yaml:"use_external_ip"`
 
-	MaxBitrate    uint64 `yaml:"max_bandwidth"`
+	MaxBitrate    uint64 `yaml:"max_bitrate"`
 	MaxBufferTime int    `yaml:"max_buffer_time"`
 }
 
@@ -44,6 +44,7 @@ func NewConfig(confString string) (*Config, error) {
 			StunServers: []string{
 				"stun.l.google.com:19302",
 			},
+			MaxBitrate: 2 * 1024 * 1024, // 3 mbps
 		},
 		Redis: RedisConfig{},
 		Keys:  map[string]string{},

pkg/rtc/mediaengine.go

@@ -0,0 +1,88 @@
+package rtc
+
+import (
+	"github.com/pion/sdp/v3"
+	"github.com/pion/webrtc/v3"
+)
+
+const (
+	repairedRTP  = "urn:ietf:params:rtp-hdrext:sdes:repaired-rtp-stream-id"
+	frameMarking = "urn:ietf:params:rtp-hdrext:framemarking"
+)
+
+func createMediaEngine() (*webrtc.MediaEngine, error) {
+	me := &webrtc.MediaEngine{}
+	if err := me.RegisterCodec(webrtc.RTPCodecParameters{
+		RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus, ClockRate: 48000, Channels: 2, SDPFmtpLine: "minptime=10;useinbandfec=1", RTCPFeedback: nil},
+		PayloadType:        111,
+	}, webrtc.RTPCodecTypeAudio); err != nil {
+		return nil, err
+	}
+
+	videoRTCPFeedback := []webrtc.RTCPFeedback{
+		{webrtc.TypeRTCPFBGoogREMB, ""},
+		//{webrtc.TypeRTCPFBCCM, "fir"},
+		{webrtc.TypeRTCPFBNACK, ""},
+		{webrtc.TypeRTCPFBNACK, "pli"}}
+	for _, codec := range []webrtc.RTPCodecParameters{
+		{
+			RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8, ClockRate: 90000, RTCPFeedback: videoRTCPFeedback},
+			PayloadType:        96,
+		},
+		{
+			RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP9, ClockRate: 90000, SDPFmtpLine: "profile-id=0", RTCPFeedback: videoRTCPFeedback},
+			PayloadType:        98,
+		},
+		{
+			RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP9, ClockRate: 90000, SDPFmtpLine: "profile-id=1", RTCPFeedback: videoRTCPFeedback},
+			PayloadType:        100,
+		},
+		{
+			RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42001f", RTCPFeedback: videoRTCPFeedback},
+			PayloadType:        102,
+		},
+		{
+			RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=0;profile-level-id=42001f", RTCPFeedback: videoRTCPFeedback},
+			PayloadType:        127,
+		},
+		{
+			RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", RTCPFeedback: videoRTCPFeedback},
+			PayloadType:        125,
+		},
+		{
+			RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=0;profile-level-id=42e01f", RTCPFeedback: videoRTCPFeedback},
+			PayloadType:        108,
+		},
+		{
+			RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=640032", RTCPFeedback: videoRTCPFeedback},
+			PayloadType:        123,
+		},
+	} {
+		if err := me.RegisterCodec(codec, webrtc.RTPCodecTypeVideo); err != nil {
+			return nil, err
+		}
+	}
+
+	for _, extension := range []string{
+		sdp.SDESMidURI,
+		sdp.SDESRTPStreamIDURI,
+		repairedRTP,
+		//sdp.TransportCCURI,
+		frameMarking,
+	} {
+		if err := me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: extension}, webrtc.RTPCodecTypeVideo); err != nil {
+			return nil, err
+		}
+	}
+	for _, extension := range []string{
+		sdp.SDESMidURI,
+		sdp.SDESRTPStreamIDURI,
+		sdp.AudioLevelURI,
+	} {
+		if err := me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: extension}, webrtc.RTPCodecTypeAudio); err != nil {
+			return nil, err
+		}
+	}
+
+	return me, nil
+}

pkg/rtc/mediatrack.go

@@ -21,7 +21,6 @@ var (
 	maxPLIFrequency = 1 * time.Second
 	feedbackTypes   = []webrtc.RTCPFeedback{
 		{webrtc.TypeRTCPFBGoogREMB, ""},
-		{webrtc.TypeRTCPFBTransportCC, ""},
 		{webrtc.TypeRTCPFBNACK, ""},
 		{webrtc.TypeRTCPFBNACK, "pli"}}
 )
@@ -44,26 +43,26 @@ type MediaTrack struct {
 	// channel to send RTCP packets to the source
 	rtcpCh chan []rtcp.Packet
 	lock   sync.RWMutex
-	// map of target participantId -> DownTrack
-	downtracks map[string]*sfu.DownTrack
-	twcc       *twcc.Responder
-	receiver   sfu.Receiver
+	// map of target participantId -> *SubscribedTrack
+	subscribedTracks map[string]*SubscribedTrack
+	twcc             *twcc.Responder
+	receiver         sfu.Receiver
 	//lastNack   int64
 	lastPLI time.Time
 }
 
 func NewMediaTrack(trackId string, pId string, rtcpCh chan []rtcp.Packet, conf ReceiverConfig, track *webrtc.TrackRemote) *MediaTrack {
 	t := &MediaTrack{
-		id:            trackId,
-		participantId: pId,
-		ssrc:          track.SSRC(),
-		streamID:      track.StreamID(),
-		kind:          ToProtoTrackKind(track.Kind()),
-		codec:         track.Codec(),
-		conf:          conf,
-		rtcpCh:        rtcpCh,
-		lock:          sync.RWMutex{},
-		downtracks:    make(map[string]*sfu.DownTrack),
+		id:               trackId,
+		participantId:    pId,
+		ssrc:             track.SSRC(),
+		streamID:         track.StreamID(),
+		kind:             ToProtoTrackKind(track.Kind()),
+		codec:            track.Codec(),
+		conf:             conf,
+		rtcpCh:           rtcpCh,
+		lock:             sync.RWMutex{},
+		subscribedTracks: make(map[string]*SubscribedTrack),
 	}
 
 	return t
@@ -89,13 +88,13 @@ func (t *MediaTrack) IsMuted() bool {
 }
 
 func (t *MediaTrack) SetMuted(muted bool) {
-	if !t.muted.TrySet(muted) {
-		return
-	}
-	// mute all of the downtracks
+	t.muted.TrySet(muted)
+
+	// mute all of the subscribedtracks
 	t.lock.RLock()
-	for _, dt := range t.downtracks {
-		dt.Mute(muted)
+	for id, st := range t.subscribedTracks {
+		logger.Debugw("setting muted", "dstParticipant", id, "muted", muted, "track", t.ID())
+		st.SetPublisherMuted(muted)
 	}
 	t.lock.RUnlock()
 }
@@ -106,20 +105,20 @@ func (t *MediaTrack) OnClose(f func()) {
 
 // subscribes participant to current remoteTrack
 // creates and add necessary forwarders and starts them
-func (t *MediaTrack) AddSubscriber(participant types.Participant) error {
-	t.lock.RLock()
-	existingDt := t.downtracks[participant.ID()]
-	t.lock.RUnlock()
+func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	existingSt := t.subscribedTracks[sub.ID()]
 
 	// don't subscribe to the same track multiple times
-	if existingDt != nil {
+	if existingSt != nil {
 		logger.Warnw("participant already subscribed to track",
-			"participant", participant.Identity(),
-			"track", existingDt.ID())
+			"sub", sub.Identity(),
+			"track", t.ID())
 		return nil
 	}
 
-	codec := t.codec
+	codec := t.receiver.Codec()
 
 	// using DownTrack from ion-sfu
 	downTrack, err := sfu.NewDownTrack(webrtc.RTPCodecCapability{
@@ -128,12 +127,13 @@ func (t *MediaTrack) AddSubscriber(participant types.Participant) error {
 		Channels:     codec.Channels,
 		SDPFmtpLine:  codec.SDPFmtpLine,
 		RTCPFeedback: feedbackTypes,
-	}, t.receiver, t.participantId)
+	}, t.receiver, sub.ID())
 	if err != nil {
 		return err
 	}
+	subTrack := NewSubscribedTrack(downTrack)
 
-	transceiver, err := participant.PeerConnection().AddTransceiverFromTrack(downTrack, webrtc.RTPTransceiverInit{
+	transceiver, err := sub.PeerConnection().AddTransceiverFromTrack(downTrack, webrtc.RTPTransceiverInit{
 		Direction: webrtc.RTPTransceiverDirectionSendrecv,
 	})
 	if err != nil {
@@ -143,21 +143,21 @@ func (t *MediaTrack) AddSubscriber(participant types.Participant) error {
 	downTrack.SetTransceiver(transceiver)
 	// when outtrack is bound, start loop to send reports
 	downTrack.OnBind(func() {
-		downTrack.Mute(t.muted.Get())
-		t.sendDownTrackBindingReports(participant.ID(), participant.RTCPChan())
+		subTrack.SetPublisherMuted(t.IsMuted())
+		t.sendDownTrackBindingReports(sub.ID(), sub.RTCPChan())
 	})
 	downTrack.OnCloseHandler(func() {
 		t.lock.Lock()
-		delete(t.downtracks, participant.ID())
+		delete(t.subscribedTracks, sub.ID())
 		t.lock.Unlock()
 
-		// ignore if the subscribing participant is not connected
-		if participant.PeerConnection().ConnectionState() == webrtc.PeerConnectionStateClosed {
+		// ignore if the subscribing sub is not connected
+		if sub.PeerConnection().ConnectionState() == webrtc.PeerConnectionStateClosed {
 			return
 		}
 
-		// if the source has been terminated, we'll need to terminate all of the downtracks
-		// however, if the dest participant has disconnected, then we can skip
+		// if the source has been terminated, we'll need to terminate all of the subscribedtracks
+		// however, if the dest sub has disconnected, then we can skip
 		sender := transceiver.Sender()
 		if sender == nil {
 			return
@@ -165,27 +165,25 @@ func (t *MediaTrack) AddSubscriber(participant types.Participant) error {
 		logger.Debugw("removing peerconnection track",
 			"track", t.id,
 			"participantId", t.participantId,
-			"destParticipant", participant.Identity())
-		if err := participant.PeerConnection().RemoveTrack(sender); err != nil {
+			"destParticipant", sub.Identity())
+		if err := sub.PeerConnection().RemoveTrack(sender); err != nil {
 			if err == webrtc.ErrConnectionClosed {
-				// participant closing, can skip removing downtracks
+				// sub closing, can skip removing subscribedtracks
 				return
 			}
 			if _, ok := err.(*rtcerr.InvalidStateError); !ok {
 				logger.Warnw("could not remove remoteTrack from forwarder",
-					"participant", participant.Identity(),
+					"sub", sub.Identity(),
 					"err", err)
 			}
 		}
 
-		participant.RemoveDownTrack(t.streamID, downTrack)
+		sub.RemoveSubscribedTrack(t.participantId, subTrack)
 	})
 
-	t.lock.Lock()
-	t.downtracks[participant.ID()] = downTrack
-	t.lock.Unlock()
+	t.subscribedTracks[sub.ID()] = subTrack
 
-	participant.AddDownTrack(t.streamID, downTrack)
+	sub.AddSubscribedTrack(t.participantId, subTrack)
 	t.receiver.AddDownTrack(downTrack, true)
 
 	return nil
@@ -203,14 +201,14 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 	if t.Kind() == livekit.TrackType_AUDIO {
 		// TODO: audio level stuff
 	} else if t.Kind() == livekit.TrackType_VIDEO {
-		if t.twcc == nil {
-			t.twcc = twcc.NewTransportWideCCResponder(uint32(track.SSRC()))
-			t.twcc.OnFeedback(func(p rtcp.RawPacket) {
-				t.rtcpCh <- []rtcp.Packet{&p}
-			})
-		}
+		//if t.twcc == nil {
+		//	t.twcc = twcc.NewTransportWideCCResponder(uint32(track.SSRC()))
+		//	t.twcc.OnFeedback(func(p rtcp.RawPacket) {
+		//		t.rtcpCh <- []rtcp.Packet{&p}
+		//	})
+		//}
 		buff.OnTransportWideCC(func(sn uint16, timeNS int64, marker bool) {
-			t.twcc.Push(sn, timeNS, marker)
+			//t.twcc.Push(sn, timeNS, marker)
 		})
 	}
 
@@ -231,21 +229,21 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 		}
 	})
 
+	t.lock.Lock()
+	defer t.lock.Unlock()
 	if t.receiver == nil {
 		// pack ID to identify all publishedTracks
 		packedId := PackTrackId(t.participantId, track.ID())
-		t.lock.Lock()
 		t.receiver = NewWrappedReceiver(sfu.NewWebRTCReceiver(receiver, track, t.participantId), packedId)
-		t.lock.Unlock()
 		t.receiver.SetRTCPCh(t.rtcpCh)
 		t.receiver.OnCloseHandler(func() {
+			t.lock.Lock()
+			defer t.lock.Unlock()
 			// source track closed
 			if t.Kind() == livekit.TrackType_AUDIO {
 				// TODO: remove audio level observer
 			}
-			t.lock.Lock()
 			t.receiver = nil
-			t.lock.Unlock()
 		})
 	}
 	t.receiver.AddUpTrack(track, buff)
@@ -262,8 +260,8 @@ func (t *MediaTrack) RemoveSubscriber(participantId string) {
 	t.lock.RLock()
 	defer t.lock.RUnlock()
 
-	if dt := t.downtracks[participantId]; dt != nil {
-		go dt.Close()
+	if subTrack := t.subscribedTracks[participantId]; subTrack != nil {
+		go subTrack.DownTrack().Close()
 	}
 }
 
@@ -271,20 +269,24 @@ func (t *MediaTrack) RemoveAllSubscribers() {
 	logger.Debugw("removing all subscribers", "track", t.id)
 	t.lock.RLock()
 	defer t.lock.RUnlock()
-	for _, dt := range t.downtracks {
-		go dt.Close()
+	for _, subTrack := range t.subscribedTracks {
+		go subTrack.DownTrack().Close()
 	}
-	t.downtracks = make(map[string]*sfu.DownTrack)
+	t.subscribedTracks = make(map[string]*SubscribedTrack)
 }
 
 func (t *MediaTrack) sendDownTrackBindingReports(participantId string, rtcpCh chan []rtcp.Packet) {
 	var sd []rtcp.SourceDescriptionChunk
 
 	t.lock.RLock()
-	dt := t.downtracks[participantId]
+	subTrack := t.subscribedTracks[participantId]
 	t.lock.RUnlock()
 
-	chunks := dt.CreateSourceDescriptionChunks()
+	if subTrack == nil {
+		return
+	}
+
+	chunks := subTrack.DownTrack().CreateSourceDescriptionChunks()
 	if chunks == nil {
 		return
 	}

pkg/rtc/participant.go

@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"github.com/bep/debounce"
-	"github.com/pion/ion-sfu/pkg/sfu"
 	"github.com/pion/rtcp"
 	"github.com/pion/webrtc/v3"
 	"github.com/pkg/errors"
@@ -33,16 +32,17 @@ const (
 )
 
 type ParticipantImpl struct {
-	id               string
-	peerConn         types.PeerConnection
-	responseSink     routing.MessageSink
-	receiverConfig   ReceiverConfig
-	isClosed         utils.AtomicFlag
-	mediaEngine      *webrtc.MediaEngine
-	identity         string
-	state            atomic.Value // livekit.ParticipantInfo_State
-	rtcpCh           chan []rtcp.Packet
-	subscribedTracks map[string][]*sfu.DownTrack
+	id             string
+	peerConn       types.PeerConnection
+	responseSink   routing.MessageSink
+	receiverConfig ReceiverConfig
+	isClosed       utils.AtomicFlag
+	mediaEngine    *webrtc.MediaEngine
+	identity       string
+	state          atomic.Value // livekit.ParticipantInfo_State
+	rtcpCh         chan []rtcp.Packet
+	// tracks the current participant is subscribed to, map of otherParticipantId => []DownTrack
+	subscribedTracks map[string][]types.SubscribedTrack
 	// publishedTracks that participant is publishing
 	publishedTracks map[string]types.PublishedTrack
 	// client intended to publish, yet to be reconciled
@@ -64,8 +64,10 @@ type ParticipantImpl struct {
 }
 
 func NewPeerConnection(conf *WebRTCConfig) (*webrtc.PeerConnection, error) {
-	me := &webrtc.MediaEngine{}
-	me.RegisterDefaultCodecs()
+	me, err := createMediaEngine()
+	if err != nil {
+		return nil, err
+	}
 	se := conf.SettingEngine
 	se.BufferFactory = bufferFactory.GetOrNew
 
@@ -76,8 +78,6 @@ func NewPeerConnection(conf *WebRTCConfig) (*webrtc.PeerConnection, error) {
 
 func NewParticipant(identity string, pc types.PeerConnection, rs routing.MessageSink, receiverConfig ReceiverConfig) (*ParticipantImpl, error) {
 	// TODO: check to ensure params are valid, id and identity can't be empty
-	me := &webrtc.MediaEngine{}
-	me.RegisterDefaultCodecs()
 
 	participant := &ParticipantImpl{
 		id:                 utils.NewGuid(utils.ParticipantPrefix),
@@ -86,12 +86,11 @@ func NewParticipant(identity string, pc types.PeerConnection, rs routing.Message
 		responseSink:       rs,
 		receiverConfig:     receiverConfig,
 		rtcpCh:             make(chan []rtcp.Packet, 50),
-		subscribedTracks:   make(map[string][]*sfu.DownTrack),
+		subscribedTracks:   make(map[string][]types.SubscribedTrack),
 		lock:               sync.RWMutex{},
 		negotiationCond:    sync.NewCond(&sync.Mutex{}),
 		publishedTracks:    make(map[string]types.PublishedTrack, 0),
 		pendingTracks:      make(map[string]*livekit.TrackInfo),
-		mediaEngine:        me,
 		debouncedNegotiate: debounce.New(negotiationFrequency),
 	}
 	participant.state.Store(livekit.ParticipantInfo_JOINING)
@@ -249,7 +248,7 @@ func (p *ParticipantImpl) Answer(sdp webrtc.SessionDescription) (answer webrtc.S
 
 	logger.Debugw("sending answer to client",
 		"participant", p.Identity(),
-	//"sdp", sdp.SDP,
+		//"sdp", sdp.SDP,
 	)
 	err = p.responseSink.WriteMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_Answer{
@@ -300,7 +299,7 @@ func (p *ParticipantImpl) HandleAnswer(sdp webrtc.SessionDescription) error {
 	}
 	logger.Debugw("setting participant answer",
 		"participant", p.Identity(),
-	//"sdp", sdp.SDP,
+		//"sdp", sdp.SDP,
 	)
 	if err := p.peerConn.SetRemoteDescription(sdp); err != nil {
 		return errors.Wrap(err, "could not set remote description")
@@ -456,23 +455,22 @@ func (p *ParticipantImpl) PeerConnection() types.PeerConnection {
 	return p.peerConn
 }
 
-func (p *ParticipantImpl) AddDownTrack(streamId string, dt *sfu.DownTrack) {
+// add a track to the participant's subscribed list
+func (p *ParticipantImpl) AddSubscribedTrack(pubId string, subTrack types.SubscribedTrack) {
+	logger.Debugw("added subscribedTrack", "srcParticipant", pubId,
+		"participant", p.Identity())
 	p.lock.Lock()
-	p.subscribedTracks[streamId] = append(p.subscribedTracks[streamId], dt)
+	p.subscribedTracks[pubId] = append(p.subscribedTracks[pubId], subTrack)
 	p.lock.Unlock()
 }
 
-func (p *ParticipantImpl) RemoveDownTrack(streamId string, dt *sfu.DownTrack) {
+// remove a track to the participant's subscribed list
+func (p *ParticipantImpl) RemoveSubscribedTrack(pubId string, subTrack types.SubscribedTrack) {
+	logger.Debugw("removed subscribedTrack", "srcParticipant", pubId,
+		"participant", p.Identity())
 	p.lock.Lock()
 	defer p.lock.Unlock()
-	tracks := p.subscribedTracks[streamId]
-	newTracks := make([]*sfu.DownTrack, 0, len(tracks))
-	for _, track := range tracks {
-		if track != dt {
-			newTracks = append(newTracks, track)
-		}
-	}
-	p.subscribedTracks[streamId] = newTracks
+	p.subscribedTracks[pubId] = funk.Without(p.subscribedTracks[pubId], subTrack).([]types.SubscribedTrack)
 }
 
 func (p *ParticipantImpl) scheduleNegotiate() {
@@ -563,6 +561,7 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
 	var mt *MediaTrack
 	var newTrack bool
 	if trk, ok := ptrack.(*MediaTrack); ok {
+		logger.Debugw("using existing mediatrack, simulcast", "rid", track.RID())
 		mt = trk
 	} else {
 		mt = NewMediaTrack(ti.Sid, p.id, p.rtcpCh, p.receiverConfig, track)
@@ -575,6 +574,16 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
 	if newTrack {
 		p.handleTrackPublished(mt)
 	}
+
+	// TODO: video tracks the current participant is subscribed to tends to freeze when the participant adds tracks
+	// to get around this, we'll trigger a resync on all tracks it's subscribed to
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+	for _, tracks := range p.subscribedTracks {
+		for _, subTrack := range tracks {
+			subTrack.Resync()
+		}
+	}
 }
 
 func (p *ParticipantImpl) onDataChannel(dc *webrtc.DataChannel) {
@@ -655,14 +664,14 @@ func (p *ParticipantImpl) downTracksRTCPWorker() {
 		var pkts []rtcp.Packet
 		var sd []rtcp.SourceDescriptionChunk
 		p.lock.RLock()
-		for _, dts := range p.subscribedTracks {
-			for _, dt := range dts {
-				sr := dt.CreateSenderReport()
-				chunks := dt.CreateSourceDescriptionChunks()
+		for _, tracks := range p.subscribedTracks {
+			for _, subTrack := range tracks {
+				sr := subTrack.DownTrack().CreateSenderReport()
+				chunks := subTrack.DownTrack().CreateSourceDescriptionChunks()
 				if sr == nil || chunks == nil {
 					continue
 				}
-				pkts = append(pkts, dt.CreateSenderReport())
+				pkts = append(pkts, sr)
 				sd = append(sd, chunks...)
 			}
 		}

pkg/rtc/subscribedtrack.go

@@ -0,0 +1,52 @@
+package rtc
+
+import (
+	"github.com/pion/ion-sfu/pkg/sfu"
+
+	"github.com/livekit/livekit-server/pkg/utils"
+)
+
+type SubscribedTrack struct {
+	dt       *sfu.DownTrack
+	subMuted utils.AtomicFlag
+	pubMuted utils.AtomicFlag
+}
+
+func NewSubscribedTrack(dt *sfu.DownTrack) *SubscribedTrack {
+	return &SubscribedTrack{
+		dt: dt,
+	}
+}
+
+func (t *SubscribedTrack) DownTrack() *sfu.DownTrack {
+	return t.dt
+}
+
+// has subscriber indicated it wants to mute this track
+func (t *SubscribedTrack) IsMuted() bool {
+	return t.subMuted.Get()
+}
+
+// set subscriber mute preference
+func (t *SubscribedTrack) SetMuted(muted bool) {
+	t.subMuted.TrySet(muted)
+	t.updateDownTrackMute()
+}
+
+func (t *SubscribedTrack) SetPublisherMuted(muted bool) {
+	t.pubMuted.TrySet(muted)
+	t.updateDownTrackMute()
+}
+
+// cause downtrack to resync and request a keyframe
+// this is sort of a hack to prevent a subscriber from having tracks freeze on the client due to missing
+// keyframes
+func (t *SubscribedTrack) Resync() {
+	t.dt.Mute(true)
+	t.updateDownTrackMute()
+}
+
+func (t *SubscribedTrack) updateDownTrackMute() {
+	muted := t.subMuted.Get() || t.pubMuted.Get()
+	t.dt.Mute(muted)
+}

pkg/rtc/types/interfaces.go

@@ -3,10 +3,8 @@ package types
 import (
 	"time"
 
-	"github.com/pion/ion-sfu/pkg/buffer"
 	"github.com/pion/ion-sfu/pkg/sfu"
 	"github.com/pion/rtcp"
-	"github.com/pion/rtp"
 	"github.com/pion/webrtc/v3"
 
 	"github.com/livekit/livekit-server/pkg/routing"
@@ -82,8 +80,8 @@ type Participant interface {
 	OnClose(func(Participant))
 
 	// package methods
-	AddDownTrack(streamId string, dt *sfu.DownTrack)
-	RemoveDownTrack(streamId string, dt *sfu.DownTrack)
+	AddSubscribedTrack(participantId string, st SubscribedTrack)
+	RemoveSubscribedTrack(participantId string, st SubscribedTrack)
 	PeerConnection() PeerConnection
 }
 
@@ -105,28 +103,13 @@ type PublishedTrack interface {
 	OnClose(func())
 }
 
-//counterfeiter:generate . Receiver
-type Receiver interface {
-	RTPChan() <-chan buffer.ExtPacket
-	GetBufferedPacket(pktBuf []byte, sn uint16, snOffset uint16) (rtp.Packet, error)
-}
-
-// DownTrack publishes data to a target participant
-// using this interface to make testing more practical
-//counterfeiter:generate . DownTrack
-type DownTrack interface {
-	ID() string
-	WriteRTP(p rtp.Packet) error
-	IsBound() bool
-	Close()
-	OnCloseHandler(fn func())
-	OnBind(fn func())
-	SSRC() uint32
-	LastSSRC() uint32
-	SnOffset() uint16
-	TsOffset() uint32
-	GetNACKSeqNo(seqNo []uint16) []uint16
-	CreateSourceDescriptionChunks() []rtcp.SourceDescriptionChunk
+//counterfeiter:generate . SubscribedTrack
+type SubscribedTrack interface {
+	DownTrack() *sfu.DownTrack
+	IsMuted() bool
+	SetMuted(muted bool)
+	SetPublisherMuted(muted bool)
+	Resync()
 }
 
 // interface for properties of webrtc.TrackRemote

pkg/rtc/types/typesfakes/fake_participant.go

@@ -244,14 +244,14 @@ type FakeParticipant struct {
 	invocationsMutex sync.RWMutex
 }
 
-func (fake *FakeParticipant) AddDownTrack(arg1 string, arg2 *sfu.DownTrack) {
+func (fake *FakeParticipant) AddSubscribedTrack(arg1 string, arg2 *sfu.DownTrack) {
 	fake.addDownTrackMutex.Lock()
 	fake.addDownTrackArgsForCall = append(fake.addDownTrackArgsForCall, struct {
 		arg1 string
 		arg2 *sfu.DownTrack
 	}{arg1, arg2})
 	stub := fake.AddDownTrackStub
-	fake.recordInvocation("AddDownTrack", []interface{}{arg1, arg2})
+	fake.recordInvocation("AddSubscribedTrack", []interface{}{arg1, arg2})
 	fake.addDownTrackMutex.Unlock()
 	if stub != nil {
 		fake.AddDownTrackStub(arg1, arg2)
@@ -1120,7 +1120,7 @@ func (fake *FakeParticipant) RemoveDownTrack(arg1 string, arg2 *sfu.DownTrack) {
 		arg2 *sfu.DownTrack
 	}{arg1, arg2})
 	stub := fake.RemoveDownTrackStub
-	fake.recordInvocation("RemoveDownTrack", []interface{}{arg1, arg2})
+	fake.recordInvocation("RemoveSubscribedTrack", []interface{}{arg1, arg2})
 	fake.removeDownTrackMutex.Unlock()
 	if stub != nil {
 		fake.RemoveDownTrackStub(arg1, arg2)