dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-28 02:54:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

tests for MediaTrack, moving interfaces to new package

Browse Source
This commit is contained in:
David Zhao
2020-12-31 23:24:45 -08:00
parent 313236e3d3
commit 0336e9d92f
23 changed files with 1210 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/cli/client/client.go
+4 -4
View File
@@ -31,7 +31,7 @@ type RTCClient struct {
iceConnected bool
paused bool
me *webrtc.MediaEngine // optional, populated only when receiving tracks
receivers []*rtc.Receiver
receivers []*rtc.ReceiverImpl
localParticipant *livekit.ParticipantInfo
// pending actions to start after connected to peer
@@ -288,10 +288,10 @@ func (c *RTCClient) ResumeLogs() {
c.paused = false
}
func (c *RTCClient) Receivers() []*rtc.Receiver {
func (c *RTCClient) Receivers() []*rtc.ReceiverImpl {
c.lock.Lock()
defer c.lock.Unlock()
return append([]*rtc.Receiver{}, c.receivers...)
return append([]*rtc.ReceiverImpl{}, c.receivers...)
}
func (c *RTCClient) SendRequest(msg *livekit.SignalRequest) error {
@@ -446,7 +446,7 @@ func (c *RTCClient) logLoop() {
}
}
func (c *RTCClient) consumeReceiver(r *rtc.Receiver) {
func (c *RTCClient) consumeReceiver(r *rtc.ReceiverImpl) {
lastUpdate := time.Time{}
peerId, trackId := rtc.UnpackTrackId(r.TrackId())
numBytes := 0
magefile.go
+1 -1
View File
@@ -30,7 +30,7 @@ var checksummer = NewChecksummer(".", goChecksumFile, ".go")
func init() {
checksummer.IgnoredPaths = []string{
"cmd/server/wire_gen.go",
"pkg/rtc/rtcfakes",
"pkg/rtc/types/typesfakes",
}
}
pkg/rtc/datatrack.go
+2 -1
View File
@@ -6,6 +6,7 @@ import (
"github.com/pion/webrtc/v3"
"github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/livekit-server/proto/livekit"
)
@@ -70,7 +71,7 @@ func (t *DataTrack) IsMuted() bool {
return false
}
func (t *DataTrack) AddSubscriber(participant Participant) error {
func (t *DataTrack) AddSubscriber(participant types.Participant) error {
label := PackDataTrackLabel(t.participantId, t.ID(), t.dataChannel.Label())
downChannel, err := participant.PeerConnection().CreateDataChannel(label, t.dataChannelOptions())
if err != nil {
pkg/rtc/forwarder.go
+5 -24
View File
@@ -10,35 +10,16 @@ import (
"github.com/pion/rtp"
"github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/sfu"
)
type PacketBuffer interface {
GetBufferedPackets(mediaSSRC uint32, snOffset uint16, tsOffset uint32, sn []uint16) []rtp.Packet
}
// a forwarder publishes data to a target remoteTrack or datachannel
// manages the RTCP loop with the target participant
type Forwarder interface {
WriteRTP(*rtp.Packet) error
Start()
Close()
CreatedAt() time.Time
Track() *sfu.DownTrack
OnClose(func(Forwarder))
}
type RTCPWriter interface {
WriteRTCP(pkts []rtcp.Packet) error
}
type SimpleForwarder struct {
ctx context.Context
cancel context.CancelFunc
sourceRtcpCh chan []rtcp.Packet // channel to write RTCP packets to source
track *sfu.DownTrack // sender track
packetBuffer PacketBuffer
packetBuffer types.PacketBuffer
lastPli time.Time
createdAt time.Time
@@ -46,10 +27,10 @@ type SimpleForwarder struct {
once sync.Once
// handlers
onClose func(forwarder Forwarder)
onClose func(forwarder types.Forwarder)
}
func NewSimpleForwarder(ctx context.Context, rtcpCh chan []rtcp.Packet, track *sfu.DownTrack, pb PacketBuffer) *SimpleForwarder {
func NewSimpleForwarder(ctx context.Context, rtcpCh chan []rtcp.Packet, track *sfu.DownTrack, pb types.PacketBuffer) *SimpleForwarder {
ctx, cancel := context.WithCancel(ctx)
f := &SimpleForwarder{
ctx: ctx,
@@ -104,7 +85,7 @@ func (f *SimpleForwarder) WriteRTP(pkt *rtp.Packet) error {
return nil
}
func (f *SimpleForwarder) OnClose(closeFunc func(Forwarder)) {
func (f *SimpleForwarder) OnClose(closeFunc func(types.Forwarder)) {
f.onClose = closeFunc
}
pkg/rtc/helper_test.go
+5 -5
View File
@@ -1,13 +1,13 @@
package rtc_test
import (
"github.com/livekit/livekit-server/pkg/rtc/rtcfakes"
"github.com/livekit/livekit-server/pkg/rtc/types/typesfakes"
"github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/livekit-server/proto/livekit"
)
func newMockParticipant(name string) *rtcfakes.FakeParticipant {
p := &rtcfakes.FakeParticipant{}
func newMockParticipant(name string) *typesfakes.FakeParticipant {
p := &typesfakes.FakeParticipant{}
p.IDReturns(utils.NewGuid(utils.ParticipantPrefix))
p.NameReturns(name)
p.StateReturns(livekit.ParticipantInfo_JOINED)
@@ -15,8 +15,8 @@ func newMockParticipant(name string) *rtcfakes.FakeParticipant {
return p
}
func newMockTrack(kind livekit.TrackInfo_Type, name string) *rtcfakes.FakePublishedTrack {
t := &rtcfakes.FakePublishedTrack{}
func newMockTrack(kind livekit.TrackInfo_Type, name string) *typesfakes.FakePublishedTrack {
t := &typesfakes.FakePublishedTrack{}
t.IDReturns(utils.NewGuid(utils.TrackPrefix))
t.KindReturns(kind)
t.StreamIDReturns(name)
pkg/rtc/mediatrack.go
+26 -24
View File
@@ -2,7 +2,6 @@ package rtc
import (
"context"
"io"
"sync"
"time"
@@ -11,6 +10,7 @@ import (
"github.com/pion/webrtc/v3/pkg/rtcerr"
"github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/sfu"
"github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/livekit-server/proto/livekit"
@@ -28,29 +28,37 @@ type MediaTrack struct {
id string
participantId string
muted bool
// source remoteTrack
remoteTrack *webrtc.TrackRemote
// duplicated properties from TrackRemote
ssrc webrtc.SSRC
streamID string // otherwise known as label
kind livekit.TrackInfo_Type
codec webrtc.RTPCodecParameters
// channel to send RTCP packets to the source
rtcpCh chan []rtcp.Packet
lock sync.RWMutex
once sync.Once
// map of target participantId -> forwarder
forwarders map[string]Forwarder
receiver *Receiver
forwarders map[string]types.Forwarder
receiver types.Receiver
//lastNack int64
lastPLI time.Time
}
func NewMediaTrack(pId string, rtcpCh chan []rtcp.Packet, track *webrtc.TrackRemote, receiver *Receiver) *MediaTrack {
func NewMediaTrack(pId string, rtcpCh chan []rtcp.Packet, track *webrtc.TrackRemote, receiver types.Receiver) *MediaTrack {
t := &MediaTrack{
ctx: context.Background(),
id: utils.NewGuid(utils.TrackPrefix),
participantId: pId,
remoteTrack: track,
ssrc: track.SSRC(),
streamID: track.StreamID(),
kind: ToProtoTrackKind(track.Kind()),
codec: track.Codec(),
rtcpCh: rtcpCh,
lock: sync.RWMutex{},
once: sync.Once{},
forwarders: make(map[string]Forwarder),
forwarders: make(map[string]types.Forwarder),
receiver: receiver,
}
@@ -70,17 +78,11 @@ func (t *MediaTrack) ID() string {
}
func (t *MediaTrack) Kind() livekit.TrackInfo_Type {
switch t.remoteTrack.Kind() {
case webrtc.RTPCodecTypeVideo:
return livekit.TrackInfo_VIDEO
case webrtc.RTPCodecTypeAudio:
return livekit.TrackInfo_AUDIO
}
panic("unsupported track kind")
return t.kind
}
func (t *MediaTrack) StreamID() string {
return t.remoteTrack.StreamID()
return t.streamID
}
func (t *MediaTrack) IsMuted() bool {
@@ -89,8 +91,8 @@ func (t *MediaTrack) IsMuted() bool {
// subscribes participant to current remoteTrack
// creates and add necessary forwarders and starts them
func (t *MediaTrack) AddSubscriber(participant Participant) error {
codec := t.remoteTrack.Codec()
func (t *MediaTrack) AddSubscriber(participant types.Participant) error {
codec := t.codec
// pack ID to identify all publishedTracks
packedId := PackTrackId(t.participantId, t.id)
@@ -101,7 +103,7 @@ func (t *MediaTrack) AddSubscriber(participant Participant) error {
Channels: codec.Channels,
SDPFmtpLine: codec.SDPFmtpLine,
RTCPFeedback: feedbackTypes,
}, packedId, t.remoteTrack.StreamID())
}, packedId, t.StreamID())
if err != nil {
return err
}
@@ -121,7 +123,7 @@ func (t *MediaTrack) AddSubscriber(participant Participant) error {
participant.AddDownTrack(t.StreamID(), outTrack)
forwarder := NewSimpleForwarder(t.ctx, t.rtcpCh, outTrack, t.receiver)
forwarder.OnClose(func(f Forwarder) {
forwarder.OnClose(func(f types.Forwarder) {
t.lock.Lock()
delete(t.forwarders, participant.ID())
t.lock.Unlock()
@@ -171,7 +173,7 @@ func (t *MediaTrack) RemoveAllSubscribers() {
for _, f := range t.forwarders {
go f.Close()
}
t.forwarders = make(map[string]Forwarder)
t.forwarders = make(map[string]types.Forwarder)
}
// forwardRTPWorker reads from the receiver and writes to each sender
@@ -183,7 +185,7 @@ func (t *MediaTrack) forwardRTPWorker() {
for {
pkt, err := t.receiver.ReadRTP()
if err == io.EOF {
if IsEOF(err) {
logger.GetLogger().Debugw("Track received EOF, closing",
"participant", t.participantId,
"track", t.id)
@@ -207,7 +209,7 @@ func (t *MediaTrack) forwardRTPWorker() {
t.lock.RLock()
for dstId, forwarder := range t.forwarders {
err := forwarder.WriteRTP(pkt)
if err == io.EOF {
if IsEOF(err) {
// this participant unsubscribed, remove it
t.RemoveSubscriber(dstId)
continue
@@ -220,7 +222,7 @@ func (t *MediaTrack) forwardRTPWorker() {
}
logger.GetLogger().Infow("keyframe required, sending PLI")
rtcpPkts := []rtcp.Packet{
&rtcp.PictureLossIndication{SenderSSRC: uint32(t.remoteTrack.SSRC()), MediaSSRC: pkt.SSRC},
&rtcp.PictureLossIndication{SenderSSRC: uint32(t.ssrc), MediaSSRC: pkt.SSRC},
}
// queue up a PLI, but don't block channel
go func() {
pkg/rtc/mediatrack_internal_test.go
+97
View File
@@ -0,0 +1,97 @@
package rtc
import (
"context"
"io"
"sync"
"testing"
"time"
"github.com/pion/rtcp"
"github.com/pion/rtp"
"github.com/pion/webrtc/v3"
"github.com/stretchr/testify/assert"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/rtc/types/typesfakes"
"github.com/livekit/livekit-server/pkg/sfu"
"github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/livekit-server/proto/livekit"
)
const (
testWaitDuration = 10 * time.Millisecond
)
func TestForwardRTP(t *testing.T) {
t.Run("ensure that forwarders are getting packets", func(t *testing.T) {
mt := newMediaTrackWithReceiver()
receiver := mt.receiver.(*typesfakes.FakeReceiver)
packet := &rtp.Packet{}
receiver.ReadRTPReturnsOnCall(0, packet, nil)
forwarder := &typesfakes.FakeForwarder{}
mt.forwarders["test"] = forwarder
mt.Start()
time.Sleep(testWaitDuration)
assert.Equal(t, 2, receiver.ReadRTPCallCount(), "worker didn't call ReadRTP twice")
assert.Equal(t, 1, forwarder.WriteRTPCallCount(), "WriteRTP wasn't called on Forwarder")
assert.Equal(t, packet, forwarder.WriteRTPArgsForCall(0))
})
t.Run("muted tracks do not forward data", func(t *testing.T) {
mt := newMediaTrackWithReceiver()
mt.muted = true
forwarder := &typesfakes.FakeForwarder{}
mt.forwarders["test"] = forwarder
mt.Start()
time.Sleep(testWaitDuration)
assert.Zero(t, forwarder.WriteRTPCallCount())
})
}
func TestMissingKeyFrames(t *testing.T) {
t.Run("PLI packet is sent when forwarder misses keyframe", func(t *testing.T) {
mt := newMediaTrackWithReceiver()
forwarder := &typesfakes.FakeForwarder{}
mt.forwarders["test"] = forwarder
forwarder.WriteRTPReturns(sfu.ErrRequiresKeyFrame)
mt.Start()
time.Sleep(testWaitDuration)
select {
case pkts := <-mt.rtcpCh:
assert.Len(t, pkts, 1, "a single RTCP packet should be returned")
assert.IsType(t, &rtcp.PictureLossIndication{}, pkts[0])
default:
t.Fatalf("did not receive RTCP packets")
}
})
}
// returns a receiver that reads a packet then returns EOF
func newMediaTrackWithReceiver() *MediaTrack {
packet := &rtp.Packet{}
receiver := &typesfakes.FakeReceiver{}
receiver.ReadRTPReturnsOnCall(0, packet, nil)
receiver.ReadRTPReturnsOnCall(1, nil, io.EOF)
return &MediaTrack{
ctx: context.Background(),
id: utils.NewGuid(utils.TrackPrefix),
participantId: "PAtest",
muted: false,
kind: livekit.TrackInfo_VIDEO,
codec: webrtc.RTPCodecParameters{},
rtcpCh: make(chan []rtcp.Packet, 5),
lock: sync.RWMutex{},
once: sync.Once{},
forwarders: map[string]types.Forwarder{},
receiver: receiver,
}
}
pkg/rtc/participant.go
+20 -18
View File
@@ -13,6 +13,7 @@ import (
"github.com/pkg/errors"
"github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/sfu"
"github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/livekit-server/proto/livekit"
@@ -25,8 +26,8 @@ const (
type ParticipantImpl struct {
id string
peerConn PeerConnection
sigConn SignalConnection
peerConn types.PeerConnection
sigConn types.SignalConnection
ctx context.Context
cancel context.CancelFunc
mediaEngine *webrtc.MediaEngine
@@ -35,18 +36,19 @@ type ParticipantImpl struct {
bi *buffer.Interceptor
rtcpCh chan []rtcp.Packet
subscribedTracks map[string][]*sfu.DownTrack
publishedTracks map[string]PublishedTrack // publishedTracks that the peer is publishing
// publishedTracks that participant is publishing
publishedTracks map[string]types.PublishedTrack
lock sync.RWMutex
once sync.Once
// callbacks & handlers
onTrackPublished func(Participant, PublishedTrack)
onTrackUpdated func(Participant, PublishedTrack)
onTrackPublished func(types.Participant, types.PublishedTrack)
onTrackUpdated func(types.Participant, types.PublishedTrack)
onOffer func(webrtc.SessionDescription)
onICECandidate func(c *webrtc.ICECandidateInit)
onStateChange func(p Participant, oldState livekit.ParticipantInfo_State)
onClose func(Participant)
onStateChange func(p types.Participant, oldState livekit.ParticipantInfo_State)
onClose func(types.Participant)
}
func NewPeerConnection(conf *WebRTCConfig) (*webrtc.PeerConnection, error) {
@@ -56,7 +58,7 @@ func NewPeerConnection(conf *WebRTCConfig) (*webrtc.PeerConnection, error) {
return api.NewPeerConnection(conf.Configuration)
}
func NewParticipant(pc PeerConnection, sc SignalConnection, name string) (*ParticipantImpl, error) {
func NewParticipant(pc types.PeerConnection, sc types.SignalConnection, name string) (*ParticipantImpl, error) {
me := &webrtc.MediaEngine{}
me.RegisterDefaultCodecs()
@@ -77,7 +79,7 @@ func NewParticipant(pc PeerConnection, sc SignalConnection, name string) (*Parti
subscribedTracks: make(map[string][]*sfu.DownTrack),
state: livekit.ParticipantInfo_JOINING,
lock: sync.RWMutex{},
publishedTracks: make(map[string]PublishedTrack, 0),
publishedTracks: make(map[string]types.PublishedTrack, 0),
mediaEngine: me,
}
@@ -146,7 +148,7 @@ func (p *ParticipantImpl) ToProto() *livekit.ParticipantInfo {
}
// callbacks for clients
func (p *ParticipantImpl) OnTrackPublished(callback func(Participant, PublishedTrack)) {
func (p *ParticipantImpl) OnTrackPublished(callback func(types.Participant, types.PublishedTrack)) {
p.onTrackPublished = callback
}
@@ -158,15 +160,15 @@ func (p *ParticipantImpl) OnICECandidate(callback func(c *webrtc.ICECandidateIni
p.onICECandidate = callback
}
func (p *ParticipantImpl) OnStateChange(callback func(p Participant, oldState livekit.ParticipantInfo_State)) {
func (p *ParticipantImpl) OnStateChange(callback func(p types.Participant, oldState livekit.ParticipantInfo_State)) {
p.onStateChange = callback
}
func (p *ParticipantImpl) OnTrackUpdated(callback func(Participant, PublishedTrack)) {
func (p *ParticipantImpl) OnTrackUpdated(callback func(types.Participant, types.PublishedTrack)) {
p.onTrackUpdated = callback
}
func (p *ParticipantImpl) OnClose(callback func(Participant)) {
func (p *ParticipantImpl) OnClose(callback func(types.Participant)) {
p.onClose = callback
}
@@ -294,7 +296,7 @@ func (p *ParticipantImpl) Close() error {
}
// Subscribes otherPeer to all of the publishedTracks
func (p *ParticipantImpl) AddSubscriber(op Participant) error {
func (p *ParticipantImpl) AddSubscriber(op types.Participant) error {
p.lock.RLock()
defer p.lock.RUnlock()
@@ -320,7 +322,7 @@ func (p *ParticipantImpl) RemoveSubscriber(participantId string) {
}
// signal connection methods
func (p *ParticipantImpl) SendJoinResponse(roomInfo *livekit.RoomInfo, otherParticipants []Participant) error {
func (p *ParticipantImpl) SendJoinResponse(roomInfo *livekit.RoomInfo, otherParticipants []types.Participant) error {
// send Join response
return p.sigConn.WriteResponse(&livekit.SignalResponse{
Message: &livekit.SignalResponse_Join{
@@ -363,7 +365,7 @@ func (p *ParticipantImpl) SetTrackMuted(trackId string, muted bool) {
}
}
func (p *ParticipantImpl) PeerConnection() PeerConnection {
func (p *ParticipantImpl) PeerConnection() types.PeerConnection {
return p.peerConn
}
@@ -407,7 +409,7 @@ func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) {
func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
logger.GetLogger().Debugw("remoteTrack added", "participantId", p.ID(), "remoteTrack", track.ID())
// create Receiver
// create ReceiverImpl
receiver := NewReceiver(p.id, rtpReceiver, p.bi)
mt := NewMediaTrack(p.id, p.rtcpCh, track, receiver)
@@ -430,7 +432,7 @@ func (p *ParticipantImpl) onDataChannel(dc *webrtc.DataChannel) {
p.handleTrackPublished(dt)
}
func (p *ParticipantImpl) handleTrackPublished(track PublishedTrack) {
func (p *ParticipantImpl) handleTrackPublished(track types.PublishedTrack) {
p.lock.Lock()
p.publishedTracks[track.ID()] = track
p.lock.Unlock()
pkg/rtc/receiver.go
+20 -24
View File
@@ -16,54 +16,50 @@ const (
)
// A receiver is responsible for pulling from a remoteTrack
type Receiver struct {
peerId string
rtpReceiver *webrtc.RTPReceiver
track *webrtc.TrackRemote
bi *buffer.Interceptor
once sync.Once
bytesRead int64
type ReceiverImpl struct {
participantId string
rtpReceiver *webrtc.RTPReceiver
track *webrtc.TrackRemote
bi *buffer.Interceptor
once sync.Once
bytesRead int64
}
func NewReceiver(peerId string, rtpReceiver *webrtc.RTPReceiver, bi *buffer.Interceptor) *Receiver {
return &Receiver{
peerId: peerId,
rtpReceiver: rtpReceiver,
track: rtpReceiver.Track(),
bi: bi,
once: sync.Once{},
func NewReceiver(peerId string, rtpReceiver *webrtc.RTPReceiver, bi *buffer.Interceptor) *ReceiverImpl {
return &ReceiverImpl{
participantId: peerId,
rtpReceiver: rtpReceiver,
track: rtpReceiver.Track(),
bi: bi,
once: sync.Once{},
}
}
func (r *Receiver) PeerId() string {
return r.peerId
}
func (r *Receiver) TrackId() string {
func (r *ReceiverImpl) TrackId() string {
return r.track.ID()
}
// starts reading RTP and push to buffer
func (r *Receiver) Start() {
func (r *ReceiverImpl) Start() {
r.once.Do(func() {
go r.rtcpWorker()
})
}
// PacketBuffer interface, to provide forwarders packets from the buffer
func (r *Receiver) GetBufferedPackets(mediaSSRC uint32, snOffset uint16, tsOffset uint32, sn []uint16) []rtp.Packet {
func (r *ReceiverImpl) GetBufferedPackets(mediaSSRC uint32, snOffset uint16, tsOffset uint32, sn []uint16) []rtp.Packet {
if r.bi == nil {
return nil
}
return r.bi.GetBufferedPackets(uint32(r.track.SSRC()), mediaSSRC, snOffset, tsOffset, sn)
}
func (r *Receiver) ReadRTP() (*rtp.Packet, error) {
func (r *ReceiverImpl) ReadRTP() (*rtp.Packet, error) {
return r.track.ReadRTP()
}
// rtcpWorker reads RTCP messages from receiver, notifies buffer
func (r *Receiver) rtcpWorker() {
func (r *ReceiverImpl) rtcpWorker() {
// consume RTCP from the sender/source, but don't need to do anything with the packets
for {
_, err := r.rtpReceiver.ReadRTCP()
@@ -72,7 +68,7 @@ func (r *Receiver) rtcpWorker() {
}
if err != nil {
logger.GetLogger().Warnw("receiver error reading RTCP",
"peer", r.peerId,
"participant", r.participantId,
"remoteTrack", r.track.SSRC(),
"err", err,
)
pkg/rtc/room.go
+13 -12
View File
@@ -7,6 +7,7 @@ import (
"github.com/thoas/go-funk"
"github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/livekit-server/proto/livekit"
)
@@ -16,7 +17,7 @@ type Room struct {
config WebRTCConfig
lock sync.RWMutex
// map of participantId -> Participant
participants map[string]Participant
participants map[string]types.Participant
}
func NewRoomForRequest(req *livekit.CreateRoomRequest, config *WebRTCConfig) *Room {
@@ -30,20 +31,20 @@ func NewRoomForRequest(req *livekit.CreateRoomRequest, config *WebRTCConfig) *Ro
},
config: *config,
lock: sync.RWMutex{},
participants: make(map[string]Participant),
participants: make(map[string]types.Participant),
}
}
func (r *Room) GetParticipant(id string) Participant {
func (r *Room) GetParticipant(id string) types.Participant {
r.lock.RLock()
defer r.lock.RUnlock()
return r.participants[id]
}
func (r *Room) GetParticipants() []Participant {
func (r *Room) GetParticipants() []types.Participant {
r.lock.RLock()
defer r.lock.RUnlock()
return funk.Values(r.participants).([]Participant)
return funk.Values(r.participants).([]types.Participant)
}
func (r *Room) ToRoomInfo(node *livekit.Node) *livekit.RoomInfo {
@@ -58,7 +59,7 @@ func (r *Room) ToRoomInfo(node *livekit.Node) *livekit.RoomInfo {
return ri
}
func (r *Room) Join(participant Participant) error {
func (r *Room) Join(participant types.Participant) error {
r.lock.Lock()
defer r.lock.Unlock()
@@ -66,7 +67,7 @@ func (r *Room) Join(participant Participant) error {
// it's important to set this before connection, we don't want to miss out on any publishedTracks
participant.OnTrackPublished(r.onTrackAdded)
participant.OnStateChange(func(p Participant, oldState livekit.ParticipantInfo_State) {
participant.OnStateChange(func(p types.Participant, oldState livekit.ParticipantInfo_State) {
log.Debugw("participant state changed", "state", p.State(), "participant", p.ID())
r.broadcastParticipantState(p)
@@ -97,7 +98,7 @@ func (r *Room) Join(participant Participant) error {
r.participants[participant.ID()] = participant
// gather other participants and send join response
otherParticipants := make([]Participant, 0, len(r.participants))
otherParticipants := make([]types.Participant, 0, len(r.participants))
for _, p := range r.participants {
if p.ID() != participant.ID() {
otherParticipants = append(otherParticipants, p)
@@ -125,7 +126,7 @@ func (r *Room) RemoveParticipant(id string) {
}
// a ParticipantImpl in the room added a new remoteTrack, subscribe other participants to it
func (r *Room) onTrackAdded(participant Participant, track PublishedTrack) {
func (r *Room) onTrackAdded(participant types.Participant, track types.PublishedTrack) {
// publish participant update, since track state is changed
r.broadcastParticipantState(participant)
@@ -152,15 +153,15 @@ func (r *Room) onTrackAdded(participant Participant, track PublishedTrack) {
}
}
func (r *Room) onTrackMuted(p Participant, track PublishedTrack) {
func (r *Room) onTrackMuted(p types.Participant, track types.PublishedTrack) {
r.broadcastParticipantState(p)
}
func (r *Room) broadcastParticipantState(p Participant) {
func (r *Room) broadcastParticipantState(p types.Participant) {
r.lock.RLock()
defer r.lock.RUnlock()
updates := ToProtoParticipants([]Participant{p})
updates := ToProtoParticipants([]types.Participant{p})
for _, op := range r.participants {
// skip itself
if p.ID() == op.ID() {
pkg/rtc/room_test.go
+8 -8
View File
@@ -7,7 +7,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/livekit/livekit-server/pkg/rtc"
"github.com/livekit/livekit-server/pkg/rtc/rtcfakes"
"github.com/livekit/livekit-server/pkg/rtc/types/typesfakes"
"github.com/livekit/livekit-server/proto/livekit"
)
@@ -73,7 +73,7 @@ func TestRoomJoin(t *testing.T) {
if p == op {
continue
}
mockP := op.(*rtcfakes.FakeParticipant)
mockP := op.(*typesfakes.FakeParticipant)
assert.NotZero(t, mockP.AddSubscriberCallCount())
// last call should be to add the newest participant
assert.Equal(t, p, mockP.AddSubscriberArgsForCall(mockP.AddSubscriberCallCount()-1))
@@ -83,7 +83,7 @@ func TestRoomJoin(t *testing.T) {
t.Run("participant removal is broadcasted to others", func(t *testing.T) {
rm := newRoomWithParticipants(t, numParticipants)
participants := rm.GetParticipants()
p := participants[0].(*rtcfakes.FakeParticipant)
p := participants[0].(*typesfakes.FakeParticipant)
rm.RemoveParticipant(p.ID())
time.Sleep(10 * time.Millisecond)
@@ -93,7 +93,7 @@ func TestRoomJoin(t *testing.T) {
assert.Zero(t, p.SendParticipantUpdateCallCount())
continue
}
fakeP := op.(*rtcfakes.FakeParticipant)
fakeP := op.(*typesfakes.FakeParticipant)
assert.Equal(t, 1, fakeP.SendParticipantUpdateCallCount())
}
})
@@ -103,13 +103,13 @@ func TestNewTrack(t *testing.T) {
t.Run("new track should be added to connected participants", func(t *testing.T) {
rm := newRoomWithParticipants(t, 4)
participants := rm.GetParticipants()
p0 := participants[0].(*rtcfakes.FakeParticipant)
p0 := participants[0].(*typesfakes.FakeParticipant)
p0.StateReturns(livekit.ParticipantInfo_JOINING)
p1 := participants[1].(*rtcfakes.FakeParticipant)
p1 := participants[1].(*typesfakes.FakeParticipant)
p1.StateReturns(livekit.ParticipantInfo_DISCONNECTED)
p2 := participants[2].(*rtcfakes.FakeParticipant)
p2 := participants[2].(*typesfakes.FakeParticipant)
p2.StateReturns(livekit.ParticipantInfo_JOINED)
p3 := participants[3].(*rtcfakes.FakeParticipant)
p3 := participants[3].(*typesfakes.FakeParticipant)
// p3 adds track
track := newMockTrack(livekit.TrackInfo_VIDEO, "webcam")
pkg/rtc/interfaces.go → pkg/rtc/types/interfaces.go
+28 -1
View File
@@ -1,9 +1,10 @@
package rtc
package types
import (
"time"
"github.com/pion/rtcp"
"github.com/pion/rtp"
"github.com/pion/webrtc/v3"
"github.com/livekit/livekit-server/pkg/sfu"
@@ -97,3 +98,29 @@ type PublishedTrack interface {
RemoveSubscriber(participantId string)
RemoveAllSubscribers()
}
//counterfeiter:generate . Receiver
type Receiver interface {
TrackId() string
Start()
GetBufferedPackets(mediaSSRC uint32, snOffset uint16, tsOffset uint32, sn []uint16) []rtp.Packet
ReadRTP() (*rtp.Packet, error)
}
//counterfeiter:generate . PacketBuffer
type PacketBuffer interface {
GetBufferedPackets(mediaSSRC uint32, snOffset uint16, tsOffset uint32, sn []uint16) []rtp.Packet
}
// a Forwarder publishes data to a target remoteTrack or datachannel
// manages the RTCP loop with the target participant
//counterfeiter:generate . Forwarder
type Forwarder interface {
WriteRTP(*rtp.Packet) error
Start()
Close()
CreatedAt() time.Time
Track() *sfu.DownTrack
OnClose(func(Forwarder))
}
pkg/rtc/types/typesfakes/fake_forwarder.go
+343
View File
@@ -0,0 +1,343 @@
// Code generated by counterfeiter. DO NOT EDIT.
package typesfakes
import (
"sync"
"time"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/sfu"
"github.com/pion/rtp"
)
type FakeForwarder struct {
CloseStub func()
closeMutex sync.RWMutex
closeArgsForCall []struct {
}
CreatedAtStub func() time.Time
createdAtMutex sync.RWMutex
createdAtArgsForCall []struct {
}
createdAtReturns struct {
result1 time.Time
}
createdAtReturnsOnCall map[int]struct {
result1 time.Time
}
OnCloseStub func(func(types.Forwarder))
onCloseMutex sync.RWMutex
onCloseArgsForCall []struct {
arg1 func(types.Forwarder)
}
StartStub func()
startMutex sync.RWMutex
startArgsForCall []struct {
}
TrackStub func() *sfu.DownTrack
trackMutex sync.RWMutex
trackArgsForCall []struct {
}
trackReturns struct {
result1 *sfu.DownTrack
}
trackReturnsOnCall map[int]struct {
result1 *sfu.DownTrack
}
WriteRTPStub func(*rtp.Packet) error
writeRTPMutex sync.RWMutex
writeRTPArgsForCall []struct {
arg1 *rtp.Packet
}
writeRTPReturns struct {
result1 error
}
writeRTPReturnsOnCall map[int]struct {
result1 error
}
invocations map[string][][]interface{}
invocationsMutex sync.RWMutex
}
func (fake *FakeForwarder) Close() {
fake.closeMutex.Lock()
fake.closeArgsForCall = append(fake.closeArgsForCall, struct {
}{})
stub := fake.CloseStub
fake.recordInvocation("Close", []interface{}{})
fake.closeMutex.Unlock()
if stub != nil {
fake.CloseStub()
}
}
func (fake *FakeForwarder) CloseCallCount() int {
fake.closeMutex.RLock()
defer fake.closeMutex.RUnlock()
return len(fake.closeArgsForCall)
}
func (fake *FakeForwarder) CloseCalls(stub func()) {
fake.closeMutex.Lock()
defer fake.closeMutex.Unlock()
fake.CloseStub = stub
}
func (fake *FakeForwarder) CreatedAt() time.Time {
fake.createdAtMutex.Lock()
ret, specificReturn := fake.createdAtReturnsOnCall[len(fake.createdAtArgsForCall)]
fake.createdAtArgsForCall = append(fake.createdAtArgsForCall, struct {
}{})
stub := fake.CreatedAtStub
fakeReturns := fake.createdAtReturns
fake.recordInvocation("CreatedAt", []interface{}{})
fake.createdAtMutex.Unlock()
if stub != nil {
return stub()
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeForwarder) CreatedAtCallCount() int {
fake.createdAtMutex.RLock()
defer fake.createdAtMutex.RUnlock()
return len(fake.createdAtArgsForCall)
}
func (fake *FakeForwarder) CreatedAtCalls(stub func() time.Time) {
fake.createdAtMutex.Lock()
defer fake.createdAtMutex.Unlock()
fake.CreatedAtStub = stub
}
func (fake *FakeForwarder) CreatedAtReturns(result1 time.Time) {
fake.createdAtMutex.Lock()
defer fake.createdAtMutex.Unlock()
fake.CreatedAtStub = nil
fake.createdAtReturns = struct {
result1 time.Time
}{result1}
}
func (fake *FakeForwarder) CreatedAtReturnsOnCall(i int, result1 time.Time) {
fake.createdAtMutex.Lock()
defer fake.createdAtMutex.Unlock()
fake.CreatedAtStub = nil
if fake.createdAtReturnsOnCall == nil {
fake.createdAtReturnsOnCall = make(map[int]struct {
result1 time.Time
})
}
fake.createdAtReturnsOnCall[i] = struct {
result1 time.Time
}{result1}
}
func (fake *FakeForwarder) OnClose(arg1 func(types.Forwarder)) {
fake.onCloseMutex.Lock()
fake.onCloseArgsForCall = append(fake.onCloseArgsForCall, struct {
arg1 func(types.Forwarder)
}{arg1})
stub := fake.OnCloseStub
fake.recordInvocation("OnClose", []interface{}{arg1})
fake.onCloseMutex.Unlock()
if stub != nil {
fake.OnCloseStub(arg1)
}
}
func (fake *FakeForwarder) OnCloseCallCount() int {
fake.onCloseMutex.RLock()
defer fake.onCloseMutex.RUnlock()
return len(fake.onCloseArgsForCall)
}
func (fake *FakeForwarder) OnCloseCalls(stub func(func(types.Forwarder))) {
fake.onCloseMutex.Lock()
defer fake.onCloseMutex.Unlock()
fake.OnCloseStub = stub
}
func (fake *FakeForwarder) OnCloseArgsForCall(i int) func(types.Forwarder) {
fake.onCloseMutex.RLock()
defer fake.onCloseMutex.RUnlock()
argsForCall := fake.onCloseArgsForCall[i]
return argsForCall.arg1
}
func (fake *FakeForwarder) Start() {
fake.startMutex.Lock()
fake.startArgsForCall = append(fake.startArgsForCall, struct {
}{})
stub := fake.StartStub
fake.recordInvocation("Start", []interface{}{})
fake.startMutex.Unlock()
if stub != nil {
fake.StartStub()
}
}
func (fake *FakeForwarder) StartCallCount() int {
fake.startMutex.RLock()
defer fake.startMutex.RUnlock()
return len(fake.startArgsForCall)
}
func (fake *FakeForwarder) StartCalls(stub func()) {
fake.startMutex.Lock()
defer fake.startMutex.Unlock()
fake.StartStub = stub
}
func (fake *FakeForwarder) Track() *sfu.DownTrack {
fake.trackMutex.Lock()
ret, specificReturn := fake.trackReturnsOnCall[len(fake.trackArgsForCall)]
fake.trackArgsForCall = append(fake.trackArgsForCall, struct {
}{})
stub := fake.TrackStub
fakeReturns := fake.trackReturns
fake.recordInvocation("Track", []interface{}{})
fake.trackMutex.Unlock()
if stub != nil {
return stub()
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeForwarder) TrackCallCount() int {
fake.trackMutex.RLock()
defer fake.trackMutex.RUnlock()
return len(fake.trackArgsForCall)
}
func (fake *FakeForwarder) TrackCalls(stub func() *sfu.DownTrack) {
fake.trackMutex.Lock()
defer fake.trackMutex.Unlock()
fake.TrackStub = stub
}
func (fake *FakeForwarder) TrackReturns(result1 *sfu.DownTrack) {
fake.trackMutex.Lock()
defer fake.trackMutex.Unlock()
fake.TrackStub = nil
fake.trackReturns = struct {
result1 *sfu.DownTrack
}{result1}
}
func (fake *FakeForwarder) TrackReturnsOnCall(i int, result1 *sfu.DownTrack) {
fake.trackMutex.Lock()
defer fake.trackMutex.Unlock()
fake.TrackStub = nil
if fake.trackReturnsOnCall == nil {
fake.trackReturnsOnCall = make(map[int]struct {
result1 *sfu.DownTrack
})
}
fake.trackReturnsOnCall[i] = struct {
result1 *sfu.DownTrack
}{result1}
}
func (fake *FakeForwarder) WriteRTP(arg1 *rtp.Packet) error {
fake.writeRTPMutex.Lock()
ret, specificReturn := fake.writeRTPReturnsOnCall[len(fake.writeRTPArgsForCall)]
fake.writeRTPArgsForCall = append(fake.writeRTPArgsForCall, struct {
arg1 *rtp.Packet
}{arg1})
stub := fake.WriteRTPStub
fakeReturns := fake.writeRTPReturns
fake.recordInvocation("WriteRTP", []interface{}{arg1})
fake.writeRTPMutex.Unlock()
if stub != nil {
return stub(arg1)
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeForwarder) WriteRTPCallCount() int {
fake.writeRTPMutex.RLock()
defer fake.writeRTPMutex.RUnlock()
return len(fake.writeRTPArgsForCall)
}
func (fake *FakeForwarder) WriteRTPCalls(stub func(*rtp.Packet) error) {
fake.writeRTPMutex.Lock()
defer fake.writeRTPMutex.Unlock()
fake.WriteRTPStub = stub
}
func (fake *FakeForwarder) WriteRTPArgsForCall(i int) *rtp.Packet {
fake.writeRTPMutex.RLock()
defer fake.writeRTPMutex.RUnlock()
argsForCall := fake.writeRTPArgsForCall[i]
return argsForCall.arg1
}
func (fake *FakeForwarder) WriteRTPReturns(result1 error) {
fake.writeRTPMutex.Lock()
defer fake.writeRTPMutex.Unlock()
fake.WriteRTPStub = nil
fake.writeRTPReturns = struct {
result1 error
}{result1}
}
func (fake *FakeForwarder) WriteRTPReturnsOnCall(i int, result1 error) {
fake.writeRTPMutex.Lock()
defer fake.writeRTPMutex.Unlock()
fake.WriteRTPStub = nil
if fake.writeRTPReturnsOnCall == nil {
fake.writeRTPReturnsOnCall = make(map[int]struct {
result1 error
})
}
fake.writeRTPReturnsOnCall[i] = struct {
result1 error
}{result1}
}
func (fake *FakeForwarder) Invocations() map[string][][]interface{} {
fake.invocationsMutex.RLock()
defer fake.invocationsMutex.RUnlock()
fake.closeMutex.RLock()
defer fake.closeMutex.RUnlock()
fake.createdAtMutex.RLock()
defer fake.createdAtMutex.RUnlock()
fake.onCloseMutex.RLock()
defer fake.onCloseMutex.RUnlock()
fake.startMutex.RLock()
defer fake.startMutex.RUnlock()
fake.trackMutex.RLock()
defer fake.trackMutex.RUnlock()
fake.writeRTPMutex.RLock()
defer fake.writeRTPMutex.RUnlock()
copiedInvocations := map[string][][]interface{}{}
for key, value := range fake.invocations {
copiedInvocations[key] = value
}
return copiedInvocations
}
func (fake *FakeForwarder) recordInvocation(key string, args []interface{}) {
fake.invocationsMutex.Lock()
defer fake.invocationsMutex.Unlock()
if fake.invocations == nil {
fake.invocations = map[string][][]interface{}{}
}
if fake.invocations[key] == nil {
fake.invocations[key] = [][]interface{}{}
}
fake.invocations[key] = append(fake.invocations[key], args)
}
var _ types.Forwarder = new(FakeForwarder)
pkg/rtc/types/typesfakes/fake_packet_buffer.go
+123
View File
@@ -0,0 +1,123 @@
// Code generated by counterfeiter. DO NOT EDIT.
package typesfakes
import (
"sync"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/pion/rtp"
)
type FakePacketBuffer struct {
GetBufferedPacketsStub func(uint32, uint16, uint32, []uint16) []rtp.Packet
getBufferedPacketsMutex sync.RWMutex
getBufferedPacketsArgsForCall []struct {
arg1 uint32
arg2 uint16
arg3 uint32
arg4 []uint16
}
getBufferedPacketsReturns struct {
result1 []rtp.Packet
}
getBufferedPacketsReturnsOnCall map[int]struct {
result1 []rtp.Packet
}
invocations map[string][][]interface{}
invocationsMutex sync.RWMutex
}
func (fake *FakePacketBuffer) GetBufferedPackets(arg1 uint32, arg2 uint16, arg3 uint32, arg4 []uint16) []rtp.Packet {
var arg4Copy []uint16
if arg4 != nil {
arg4Copy = make([]uint16, len(arg4))
copy(arg4Copy, arg4)
}
fake.getBufferedPacketsMutex.Lock()
ret, specificReturn := fake.getBufferedPacketsReturnsOnCall[len(fake.getBufferedPacketsArgsForCall)]
fake.getBufferedPacketsArgsForCall = append(fake.getBufferedPacketsArgsForCall, struct {
arg1 uint32
arg2 uint16
arg3 uint32
arg4 []uint16
}{arg1, arg2, arg3, arg4Copy})
stub := fake.GetBufferedPacketsStub
fakeReturns := fake.getBufferedPacketsReturns
fake.recordInvocation("GetBufferedPackets", []interface{}{arg1, arg2, arg3, arg4Copy})
fake.getBufferedPacketsMutex.Unlock()
if stub != nil {
return stub(arg1, arg2, arg3, arg4)
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakePacketBuffer) GetBufferedPacketsCallCount() int {
fake.getBufferedPacketsMutex.RLock()
defer fake.getBufferedPacketsMutex.RUnlock()
return len(fake.getBufferedPacketsArgsForCall)
}
func (fake *FakePacketBuffer) GetBufferedPacketsCalls(stub func(uint32, uint16, uint32, []uint16) []rtp.Packet) {
fake.getBufferedPacketsMutex.Lock()
defer fake.getBufferedPacketsMutex.Unlock()
fake.GetBufferedPacketsStub = stub
}
func (fake *FakePacketBuffer) GetBufferedPacketsArgsForCall(i int) (uint32, uint16, uint32, []uint16) {
fake.getBufferedPacketsMutex.RLock()
defer fake.getBufferedPacketsMutex.RUnlock()
argsForCall := fake.getBufferedPacketsArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4
}
func (fake *FakePacketBuffer) GetBufferedPacketsReturns(result1 []rtp.Packet) {
fake.getBufferedPacketsMutex.Lock()
defer fake.getBufferedPacketsMutex.Unlock()
fake.GetBufferedPacketsStub = nil
fake.getBufferedPacketsReturns = struct {
result1 []rtp.Packet
}{result1}
}
func (fake *FakePacketBuffer) GetBufferedPacketsReturnsOnCall(i int, result1 []rtp.Packet) {
fake.getBufferedPacketsMutex.Lock()
defer fake.getBufferedPacketsMutex.Unlock()
fake.GetBufferedPacketsStub = nil
if fake.getBufferedPacketsReturnsOnCall == nil {
fake.getBufferedPacketsReturnsOnCall = make(map[int]struct {
result1 []rtp.Packet
})
}
fake.getBufferedPacketsReturnsOnCall[i] = struct {
result1 []rtp.Packet
}{result1}
}
func (fake *FakePacketBuffer) Invocations() map[string][][]interface{} {
fake.invocationsMutex.RLock()
defer fake.invocationsMutex.RUnlock()
fake.getBufferedPacketsMutex.RLock()
defer fake.getBufferedPacketsMutex.RUnlock()
copiedInvocations := map[string][][]interface{}{}
for key, value := range fake.invocations {
copiedInvocations[key] = value
}
return copiedInvocations
}
func (fake *FakePacketBuffer) recordInvocation(key string, args []interface{}) {
fake.invocationsMutex.Lock()
defer fake.invocationsMutex.Unlock()
if fake.invocations == nil {
fake.invocations = map[string][][]interface{}{}
}
if fake.invocations[key] == nil {
fake.invocations[key] = [][]interface{}{}
}
fake.invocations[key] = append(fake.invocations[key], args)
}
var _ types.PacketBuffer = new(FakePacketBuffer)
pkg/rtc/rtcfakes/fake_participant.go → pkg/rtc/types/typesfakes/fake_participant.go
+125 -45
View File
@@ -1,10 +1,10 @@
// Code generated by counterfeiter. DO NOT EDIT.
package rtcfakes
package typesfakes
import (
"sync"
"github.com/livekit/livekit-server/pkg/rtc"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/sfu"
"github.com/livekit/livekit-server/proto/livekit"
webrtc "github.com/pion/webrtc/v3"
@@ -28,10 +28,10 @@ type FakeParticipant struct {
addICECandidateReturnsOnCall map[int]struct {
result1 error
}
AddSubscriberStub func(rtc.Participant) error
AddSubscriberStub func(types.Participant) error
addSubscriberMutex sync.RWMutex
addSubscriberArgsForCall []struct {
arg1 rtc.Participant
arg1 types.Participant
}
addSubscriberReturns struct {
result1 error
@@ -93,10 +93,10 @@ type FakeParticipant struct {
nameReturnsOnCall map[int]struct {
result1 string
}
OnCloseStub func(func(rtc.Participant))
OnCloseStub func(func(types.Participant))
onCloseMutex sync.RWMutex
onCloseArgsForCall []struct {
arg1 func(rtc.Participant)
arg1 func(types.Participant)
}
OnICECandidateStub func(func(c *webrtc.ICECandidateInit))
onICECandidateMutex sync.RWMutex
@@ -108,25 +108,30 @@ type FakeParticipant struct {
onOfferArgsForCall []struct {
arg1 func(webrtc.SessionDescription)
}
OnStateChangeStub func(func(p rtc.Participant, oldState livekit.ParticipantInfo_State))
OnStateChangeStub func(func(p types.Participant, oldState livekit.ParticipantInfo_State))
onStateChangeMutex sync.RWMutex
onStateChangeArgsForCall []struct {
arg1 func(p rtc.Participant, oldState livekit.ParticipantInfo_State)
arg1 func(p types.Participant, oldState livekit.ParticipantInfo_State)
}
OnTrackPublishedStub func(func(rtc.Participant, rtc.PublishedTrack))
OnTrackPublishedStub func(func(types.Participant, types.PublishedTrack))
onTrackPublishedMutex sync.RWMutex
onTrackPublishedArgsForCall []struct {
arg1 func(rtc.Participant, rtc.PublishedTrack)
arg1 func(types.Participant, types.PublishedTrack)
}
PeerConnectionStub func() rtc.PeerConnection
OnTrackUpdatedStub func(func(types.Participant, types.PublishedTrack))
onTrackUpdatedMutex sync.RWMutex
onTrackUpdatedArgsForCall []struct {
arg1 func(types.Participant, types.PublishedTrack)
}
PeerConnectionStub func() types.PeerConnection
peerConnectionMutex sync.RWMutex
peerConnectionArgsForCall []struct {
}
peerConnectionReturns struct {
result1 rtc.PeerConnection
result1 types.PeerConnection
}
peerConnectionReturnsOnCall map[int]struct {
result1 rtc.PeerConnection
result1 types.PeerConnection
}
RemoveDownTrackStub func(string, *sfu.DownTrack)
removeDownTrackMutex sync.RWMutex
@@ -139,11 +144,11 @@ type FakeParticipant struct {
removeSubscriberArgsForCall []struct {
arg1 string
}
SendJoinResponseStub func(*livekit.RoomInfo, []rtc.Participant) error
SendJoinResponseStub func(*livekit.RoomInfo, []types.Participant) error
sendJoinResponseMutex sync.RWMutex
sendJoinResponseArgsForCall []struct {
arg1 *livekit.RoomInfo
arg2 []rtc.Participant
arg2 []types.Participant
}
sendJoinResponseReturns struct {
result1 error
@@ -173,6 +178,12 @@ type FakeParticipant struct {
setRemoteDescriptionReturnsOnCall map[int]struct {
result1 error
}
SetTrackMutedStub func(string, bool)
setTrackMutedMutex sync.RWMutex
setTrackMutedArgsForCall []struct {
arg1 string
arg2 bool
}
StartStub func()
startMutex sync.RWMutex
startArgsForCall []struct {
@@ -295,11 +306,11 @@ func (fake *FakeParticipant) AddICECandidateReturnsOnCall(i int, result1 error)
}{result1}
}
func (fake *FakeParticipant) AddSubscriber(arg1 rtc.Participant) error {
func (fake *FakeParticipant) AddSubscriber(arg1 types.Participant) error {
fake.addSubscriberMutex.Lock()
ret, specificReturn := fake.addSubscriberReturnsOnCall[len(fake.addSubscriberArgsForCall)]
fake.addSubscriberArgsForCall = append(fake.addSubscriberArgsForCall, struct {
arg1 rtc.Participant
arg1 types.Participant
}{arg1})
stub := fake.AddSubscriberStub
fakeReturns := fake.addSubscriberReturns
@@ -320,13 +331,13 @@ func (fake *FakeParticipant) AddSubscriberCallCount() int {
return len(fake.addSubscriberArgsForCall)
}
func (fake *FakeParticipant) AddSubscriberCalls(stub func(rtc.Participant) error) {
func (fake *FakeParticipant) AddSubscriberCalls(stub func(types.Participant) error) {
fake.addSubscriberMutex.Lock()
defer fake.addSubscriberMutex.Unlock()
fake.AddSubscriberStub = stub
}
func (fake *FakeParticipant) AddSubscriberArgsForCall(i int) rtc.Participant {
func (fake *FakeParticipant) AddSubscriberArgsForCall(i int) types.Participant {
fake.addSubscriberMutex.RLock()
defer fake.addSubscriberMutex.RUnlock()
argsForCall := fake.addSubscriberArgsForCall[i]
@@ -640,10 +651,10 @@ func (fake *FakeParticipant) NameReturnsOnCall(i int, result1 string) {
}{result1}
}
func (fake *FakeParticipant) OnClose(arg1 func(rtc.Participant)) {
func (fake *FakeParticipant) OnClose(arg1 func(types.Participant)) {
fake.onCloseMutex.Lock()
fake.onCloseArgsForCall = append(fake.onCloseArgsForCall, struct {
arg1 func(rtc.Participant)
arg1 func(types.Participant)
}{arg1})
stub := fake.OnCloseStub
fake.recordInvocation("OnClose", []interface{}{arg1})
@@ -659,13 +670,13 @@ func (fake *FakeParticipant) OnCloseCallCount() int {
return len(fake.onCloseArgsForCall)
}
func (fake *FakeParticipant) OnCloseCalls(stub func(func(rtc.Participant))) {
func (fake *FakeParticipant) OnCloseCalls(stub func(func(types.Participant))) {
fake.onCloseMutex.Lock()
defer fake.onCloseMutex.Unlock()
fake.OnCloseStub = stub
}
func (fake *FakeParticipant) OnCloseArgsForCall(i int) func(rtc.Participant) {
func (fake *FakeParticipant) OnCloseArgsForCall(i int) func(types.Participant) {
fake.onCloseMutex.RLock()
defer fake.onCloseMutex.RUnlock()
argsForCall := fake.onCloseArgsForCall[i]
@@ -736,10 +747,10 @@ func (fake *FakeParticipant) OnOfferArgsForCall(i int) func(webrtc.SessionDescri
return argsForCall.arg1
}
func (fake *FakeParticipant) OnStateChange(arg1 func(p rtc.Participant, oldState livekit.ParticipantInfo_State)) {
func (fake *FakeParticipant) OnStateChange(arg1 func(p types.Participant, oldState livekit.ParticipantInfo_State)) {
fake.onStateChangeMutex.Lock()
fake.onStateChangeArgsForCall = append(fake.onStateChangeArgsForCall, struct {
arg1 func(p rtc.Participant, oldState livekit.ParticipantInfo_State)
arg1 func(p types.Participant, oldState livekit.ParticipantInfo_State)
}{arg1})
stub := fake.OnStateChangeStub
fake.recordInvocation("OnStateChange", []interface{}{arg1})
@@ -755,23 +766,23 @@ func (fake *FakeParticipant) OnStateChangeCallCount() int {
return len(fake.onStateChangeArgsForCall)
}
func (fake *FakeParticipant) OnStateChangeCalls(stub func(func(p rtc.Participant, oldState livekit.ParticipantInfo_State))) {
func (fake *FakeParticipant) OnStateChangeCalls(stub func(func(p types.Participant, oldState livekit.ParticipantInfo_State))) {
fake.onStateChangeMutex.Lock()
defer fake.onStateChangeMutex.Unlock()
fake.OnStateChangeStub = stub
}
func (fake *FakeParticipant) OnStateChangeArgsForCall(i int) func(p rtc.Participant, oldState livekit.ParticipantInfo_State) {
func (fake *FakeParticipant) OnStateChangeArgsForCall(i int) func(p types.Participant, oldState livekit.ParticipantInfo_State) {
fake.onStateChangeMutex.RLock()
defer fake.onStateChangeMutex.RUnlock()
argsForCall := fake.onStateChangeArgsForCall[i]
return argsForCall.arg1
}
func (fake *FakeParticipant) OnTrackPublished(arg1 func(rtc.Participant, rtc.PublishedTrack)) {
func (fake *FakeParticipant) OnTrackPublished(arg1 func(types.Participant, types.PublishedTrack)) {
fake.onTrackPublishedMutex.Lock()
fake.onTrackPublishedArgsForCall = append(fake.onTrackPublishedArgsForCall, struct {
arg1 func(rtc.Participant, rtc.PublishedTrack)
arg1 func(types.Participant, types.PublishedTrack)
}{arg1})
stub := fake.OnTrackPublishedStub
fake.recordInvocation("OnTrackPublished", []interface{}{arg1})
@@ -787,20 +798,52 @@ func (fake *FakeParticipant) OnTrackPublishedCallCount() int {
return len(fake.onTrackPublishedArgsForCall)
}
func (fake *FakeParticipant) OnTrackPublishedCalls(stub func(func(rtc.Participant, rtc.PublishedTrack))) {
func (fake *FakeParticipant) OnTrackPublishedCalls(stub func(func(types.Participant, types.PublishedTrack))) {
fake.onTrackPublishedMutex.Lock()
defer fake.onTrackPublishedMutex.Unlock()
fake.OnTrackPublishedStub = stub
}
func (fake *FakeParticipant) OnTrackPublishedArgsForCall(i int) func(rtc.Participant, rtc.PublishedTrack) {
func (fake *FakeParticipant) OnTrackPublishedArgsForCall(i int) func(types.Participant, types.PublishedTrack) {
fake.onTrackPublishedMutex.RLock()
defer fake.onTrackPublishedMutex.RUnlock()
argsForCall := fake.onTrackPublishedArgsForCall[i]
return argsForCall.arg1
}
func (fake *FakeParticipant) PeerConnection() rtc.PeerConnection {
func (fake *FakeParticipant) OnTrackUpdated(arg1 func(types.Participant, types.PublishedTrack)) {
fake.onTrackUpdatedMutex.Lock()
fake.onTrackUpdatedArgsForCall = append(fake.onTrackUpdatedArgsForCall, struct {
arg1 func(types.Participant, types.PublishedTrack)
}{arg1})
stub := fake.OnTrackUpdatedStub
fake.recordInvocation("OnTrackUpdated", []interface{}{arg1})
fake.onTrackUpdatedMutex.Unlock()
if stub != nil {
fake.OnTrackUpdatedStub(arg1)
}
}
func (fake *FakeParticipant) OnTrackUpdatedCallCount() int {
fake.onTrackUpdatedMutex.RLock()
defer fake.onTrackUpdatedMutex.RUnlock()
return len(fake.onTrackUpdatedArgsForCall)
}
func (fake *FakeParticipant) OnTrackUpdatedCalls(stub func(func(types.Participant, types.PublishedTrack))) {
fake.onTrackUpdatedMutex.Lock()
defer fake.onTrackUpdatedMutex.Unlock()
fake.OnTrackUpdatedStub = stub
}
func (fake *FakeParticipant) OnTrackUpdatedArgsForCall(i int) func(types.Participant, types.PublishedTrack) {
fake.onTrackUpdatedMutex.RLock()
defer fake.onTrackUpdatedMutex.RUnlock()
argsForCall := fake.onTrackUpdatedArgsForCall[i]
return argsForCall.arg1
}
func (fake *FakeParticipant) PeerConnection() types.PeerConnection {
fake.peerConnectionMutex.Lock()
ret, specificReturn := fake.peerConnectionReturnsOnCall[len(fake.peerConnectionArgsForCall)]
fake.peerConnectionArgsForCall = append(fake.peerConnectionArgsForCall, struct {
@@ -824,32 +867,32 @@ func (fake *FakeParticipant) PeerConnectionCallCount() int {
return len(fake.peerConnectionArgsForCall)
}
func (fake *FakeParticipant) PeerConnectionCalls(stub func() rtc.PeerConnection) {
func (fake *FakeParticipant) PeerConnectionCalls(stub func() types.PeerConnection) {
fake.peerConnectionMutex.Lock()
defer fake.peerConnectionMutex.Unlock()
fake.PeerConnectionStub = stub
}
func (fake *FakeParticipant) PeerConnectionReturns(result1 rtc.PeerConnection) {
func (fake *FakeParticipant) PeerConnectionReturns(result1 types.PeerConnection) {
fake.peerConnectionMutex.Lock()
defer fake.peerConnectionMutex.Unlock()
fake.PeerConnectionStub = nil
fake.peerConnectionReturns = struct {
result1 rtc.PeerConnection
result1 types.PeerConnection
}{result1}
}
func (fake *FakeParticipant) PeerConnectionReturnsOnCall(i int, result1 rtc.PeerConnection) {
func (fake *FakeParticipant) PeerConnectionReturnsOnCall(i int, result1 types.PeerConnection) {
fake.peerConnectionMutex.Lock()
defer fake.peerConnectionMutex.Unlock()
fake.PeerConnectionStub = nil
if fake.peerConnectionReturnsOnCall == nil {
fake.peerConnectionReturnsOnCall = make(map[int]struct {
result1 rtc.PeerConnection
result1 types.PeerConnection
})
}
fake.peerConnectionReturnsOnCall[i] = struct {
result1 rtc.PeerConnection
result1 types.PeerConnection
}{result1}
}
@@ -918,17 +961,17 @@ func (fake *FakeParticipant) RemoveSubscriberArgsForCall(i int) string {
return argsForCall.arg1
}
func (fake *FakeParticipant) SendJoinResponse(arg1 *livekit.RoomInfo, arg2 []rtc.Participant) error {
var arg2Copy []rtc.Participant
func (fake *FakeParticipant) SendJoinResponse(arg1 *livekit.RoomInfo, arg2 []types.Participant) error {
var arg2Copy []types.Participant
if arg2 != nil {
arg2Copy = make([]rtc.Participant, len(arg2))
arg2Copy = make([]types.Participant, len(arg2))
copy(arg2Copy, arg2)
}
fake.sendJoinResponseMutex.Lock()
ret, specificReturn := fake.sendJoinResponseReturnsOnCall[len(fake.sendJoinResponseArgsForCall)]
fake.sendJoinResponseArgsForCall = append(fake.sendJoinResponseArgsForCall, struct {
arg1 *livekit.RoomInfo
arg2 []rtc.Participant
arg2 []types.Participant
}{arg1, arg2Copy})
stub := fake.SendJoinResponseStub
fakeReturns := fake.sendJoinResponseReturns
@@ -949,13 +992,13 @@ func (fake *FakeParticipant) SendJoinResponseCallCount() int {
return len(fake.sendJoinResponseArgsForCall)
}
func (fake *FakeParticipant) SendJoinResponseCalls(stub func(*livekit.RoomInfo, []rtc.Participant) error) {
func (fake *FakeParticipant) SendJoinResponseCalls(stub func(*livekit.RoomInfo, []types.Participant) error) {
fake.sendJoinResponseMutex.Lock()
defer fake.sendJoinResponseMutex.Unlock()
fake.SendJoinResponseStub = stub
}
func (fake *FakeParticipant) SendJoinResponseArgsForCall(i int) (*livekit.RoomInfo, []rtc.Participant) {
func (fake *FakeParticipant) SendJoinResponseArgsForCall(i int) (*livekit.RoomInfo, []types.Participant) {
fake.sendJoinResponseMutex.RLock()
defer fake.sendJoinResponseMutex.RUnlock()
argsForCall := fake.sendJoinResponseArgsForCall[i]
@@ -1112,6 +1155,39 @@ func (fake *FakeParticipant) SetRemoteDescriptionReturnsOnCall(i int, result1 er
}{result1}
}
func (fake *FakeParticipant) SetTrackMuted(arg1 string, arg2 bool) {
fake.setTrackMutedMutex.Lock()
fake.setTrackMutedArgsForCall = append(fake.setTrackMutedArgsForCall, struct {
arg1 string
arg2 bool
}{arg1, arg2})
stub := fake.SetTrackMutedStub
fake.recordInvocation("SetTrackMuted", []interface{}{arg1, arg2})
fake.setTrackMutedMutex.Unlock()
if stub != nil {
fake.SetTrackMutedStub(arg1, arg2)
}
}
func (fake *FakeParticipant) SetTrackMutedCallCount() int {
fake.setTrackMutedMutex.RLock()
defer fake.setTrackMutedMutex.RUnlock()
return len(fake.setTrackMutedArgsForCall)
}
func (fake *FakeParticipant) SetTrackMutedCalls(stub func(string, bool)) {
fake.setTrackMutedMutex.Lock()
defer fake.setTrackMutedMutex.Unlock()
fake.SetTrackMutedStub = stub
}
func (fake *FakeParticipant) SetTrackMutedArgsForCall(i int) (string, bool) {
fake.setTrackMutedMutex.RLock()
defer fake.setTrackMutedMutex.RUnlock()
argsForCall := fake.setTrackMutedArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2
}
func (fake *FakeParticipant) Start() {
fake.startMutex.Lock()
fake.startArgsForCall = append(fake.startArgsForCall, struct {
@@ -1271,6 +1347,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} {
defer fake.onStateChangeMutex.RUnlock()
fake.onTrackPublishedMutex.RLock()
defer fake.onTrackPublishedMutex.RUnlock()
fake.onTrackUpdatedMutex.RLock()
defer fake.onTrackUpdatedMutex.RUnlock()
fake.peerConnectionMutex.RLock()
defer fake.peerConnectionMutex.RUnlock()
fake.removeDownTrackMutex.RLock()
@@ -1283,6 +1361,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} {
defer fake.sendParticipantUpdateMutex.RUnlock()
fake.setRemoteDescriptionMutex.RLock()
defer fake.setRemoteDescriptionMutex.RUnlock()
fake.setTrackMutedMutex.RLock()
defer fake.setTrackMutedMutex.RUnlock()
fake.startMutex.RLock()
defer fake.startMutex.RUnlock()
fake.stateMutex.RLock()
@@ -1308,4 +1388,4 @@ func (fake *FakeParticipant) recordInvocation(key string, args []interface{}) {
fake.invocations[key] = append(fake.invocations[key], args)
}
var _ rtc.Participant = new(FakeParticipant)
var _ types.Participant = new(FakeParticipant)
pkg/rtc/rtcfakes/fake_peer_connection.go → pkg/rtc/types/typesfakes/fake_peer_connection.go
+3 -3
View File
@@ -1,10 +1,10 @@
// Code generated by counterfeiter. DO NOT EDIT.
package rtcfakes
package typesfakes
import (
"sync"
"github.com/livekit/livekit-server/pkg/rtc"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/pion/rtcp"
webrtc "github.com/pion/webrtc/v3"
)
@@ -1056,4 +1056,4 @@ func (fake *FakePeerConnection) recordInvocation(key string, args []interface{})
fake.invocations[key] = append(fake.invocations[key], args)
}
var _ rtc.PeerConnection = new(FakePeerConnection)
var _ types.PeerConnection = new(FakePeerConnection)
pkg/rtc/rtcfakes/fake_published_track.go → pkg/rtc/types/typesfakes/fake_published_track.go
+74 -9
View File
@@ -1,18 +1,18 @@
// Code generated by counterfeiter. DO NOT EDIT.
package rtcfakes
package typesfakes
import (
"sync"
"github.com/livekit/livekit-server/pkg/rtc"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/proto/livekit"
)
type FakePublishedTrack struct {
AddSubscriberStub func(rtc.Participant) error
AddSubscriberStub func(types.Participant) error
addSubscriberMutex sync.RWMutex
addSubscriberArgsForCall []struct {
arg1 rtc.Participant
arg1 types.Participant
}
addSubscriberReturns struct {
result1 error
@@ -30,6 +30,16 @@ type FakePublishedTrack struct {
iDReturnsOnCall map[int]struct {
result1 string
}
IsMutedStub func() bool
isMutedMutex sync.RWMutex
isMutedArgsForCall []struct {
}
isMutedReturns struct {
result1 bool
}
isMutedReturnsOnCall map[int]struct {
result1 bool
}
KindStub func() livekit.TrackInfo_Type
kindMutex sync.RWMutex
kindArgsForCall []struct {
@@ -67,11 +77,11 @@ type FakePublishedTrack struct {
invocationsMutex sync.RWMutex
}
func (fake *FakePublishedTrack) AddSubscriber(arg1 rtc.Participant) error {
func (fake *FakePublishedTrack) AddSubscriber(arg1 types.Participant) error {
fake.addSubscriberMutex.Lock()
ret, specificReturn := fake.addSubscriberReturnsOnCall[len(fake.addSubscriberArgsForCall)]
fake.addSubscriberArgsForCall = append(fake.addSubscriberArgsForCall, struct {
arg1 rtc.Participant
arg1 types.Participant
}{arg1})
stub := fake.AddSubscriberStub
fakeReturns := fake.addSubscriberReturns
@@ -92,13 +102,13 @@ func (fake *FakePublishedTrack) AddSubscriberCallCount() int {
return len(fake.addSubscriberArgsForCall)
}
func (fake *FakePublishedTrack) AddSubscriberCalls(stub func(rtc.Participant) error) {
func (fake *FakePublishedTrack) AddSubscriberCalls(stub func(types.Participant) error) {
fake.addSubscriberMutex.Lock()
defer fake.addSubscriberMutex.Unlock()
fake.AddSubscriberStub = stub
}
func (fake *FakePublishedTrack) AddSubscriberArgsForCall(i int) rtc.Participant {
func (fake *FakePublishedTrack) AddSubscriberArgsForCall(i int) types.Participant {
fake.addSubscriberMutex.RLock()
defer fake.addSubscriberMutex.RUnlock()
argsForCall := fake.addSubscriberArgsForCall[i]
@@ -181,6 +191,59 @@ func (fake *FakePublishedTrack) IDReturnsOnCall(i int, result1 string) {
}{result1}
}
func (fake *FakePublishedTrack) IsMuted() bool {
fake.isMutedMutex.Lock()
ret, specificReturn := fake.isMutedReturnsOnCall[len(fake.isMutedArgsForCall)]
fake.isMutedArgsForCall = append(fake.isMutedArgsForCall, struct {
}{})
stub := fake.IsMutedStub
fakeReturns := fake.isMutedReturns
fake.recordInvocation("IsMuted", []interface{}{})
fake.isMutedMutex.Unlock()
if stub != nil {
return stub()
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakePublishedTrack) IsMutedCallCount() int {
fake.isMutedMutex.RLock()
defer fake.isMutedMutex.RUnlock()
return len(fake.isMutedArgsForCall)
}
func (fake *FakePublishedTrack) IsMutedCalls(stub func() bool) {
fake.isMutedMutex.Lock()
defer fake.isMutedMutex.Unlock()
fake.IsMutedStub = stub
}
func (fake *FakePublishedTrack) IsMutedReturns(result1 bool) {
fake.isMutedMutex.Lock()
defer fake.isMutedMutex.Unlock()
fake.IsMutedStub = nil
fake.isMutedReturns = struct {
result1 bool
}{result1}
}
func (fake *FakePublishedTrack) IsMutedReturnsOnCall(i int, result1 bool) {
fake.isMutedMutex.Lock()
defer fake.isMutedMutex.Unlock()
fake.IsMutedStub = nil
if fake.isMutedReturnsOnCall == nil {
fake.isMutedReturnsOnCall = make(map[int]struct {
result1 bool
})
}
fake.isMutedReturnsOnCall[i] = struct {
result1 bool
}{result1}
}
func (fake *FakePublishedTrack) Kind() livekit.TrackInfo_Type {
fake.kindMutex.Lock()
ret, specificReturn := fake.kindReturnsOnCall[len(fake.kindArgsForCall)]
@@ -374,6 +437,8 @@ func (fake *FakePublishedTrack) Invocations() map[string][][]interface{} {
defer fake.addSubscriberMutex.RUnlock()
fake.iDMutex.RLock()
defer fake.iDMutex.RUnlock()
fake.isMutedMutex.RLock()
defer fake.isMutedMutex.RUnlock()
fake.kindMutex.RLock()
defer fake.kindMutex.RUnlock()
fake.removeAllSubscribersMutex.RLock()
@@ -403,4 +468,4 @@ func (fake *FakePublishedTrack) recordInvocation(key string, args []interface{})
fake.invocations[key] = append(fake.invocations[key], args)
}
var _ rtc.PublishedTrack = new(FakePublishedTrack)
var _ types.PublishedTrack = new(FakePublishedTrack)
pkg/rtc/types/typesfakes/fake_receiver.go
+288
View File
@@ -0,0 +1,288 @@
// Code generated by counterfeiter. DO NOT EDIT.
package typesfakes
import (
"sync"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/pion/rtp"
)
type FakeReceiver struct {
GetBufferedPacketsStub func(uint32, uint16, uint32, []uint16) []rtp.Packet
getBufferedPacketsMutex sync.RWMutex
getBufferedPacketsArgsForCall []struct {
arg1 uint32
arg2 uint16
arg3 uint32
arg4 []uint16
}
getBufferedPacketsReturns struct {
result1 []rtp.Packet
}
getBufferedPacketsReturnsOnCall map[int]struct {
result1 []rtp.Packet
}
ReadRTPStub func() (*rtp.Packet, error)
readRTPMutex sync.RWMutex
readRTPArgsForCall []struct {
}
readRTPReturns struct {
result1 *rtp.Packet
result2 error
}
readRTPReturnsOnCall map[int]struct {
result1 *rtp.Packet
result2 error
}
StartStub func()
startMutex sync.RWMutex
startArgsForCall []struct {
}
TrackIdStub func() string
trackIdMutex sync.RWMutex
trackIdArgsForCall []struct {
}
trackIdReturns struct {
result1 string
}
trackIdReturnsOnCall map[int]struct {
result1 string
}
invocations map[string][][]interface{}
invocationsMutex sync.RWMutex
}
func (fake *FakeReceiver) GetBufferedPackets(arg1 uint32, arg2 uint16, arg3 uint32, arg4 []uint16) []rtp.Packet {
var arg4Copy []uint16
if arg4 != nil {
arg4Copy = make([]uint16, len(arg4))
copy(arg4Copy, arg4)
}
fake.getBufferedPacketsMutex.Lock()
ret, specificReturn := fake.getBufferedPacketsReturnsOnCall[len(fake.getBufferedPacketsArgsForCall)]
fake.getBufferedPacketsArgsForCall = append(fake.getBufferedPacketsArgsForCall, struct {
arg1 uint32
arg2 uint16
arg3 uint32
arg4 []uint16
}{arg1, arg2, arg3, arg4Copy})
stub := fake.GetBufferedPacketsStub
fakeReturns := fake.getBufferedPacketsReturns
fake.recordInvocation("GetBufferedPackets", []interface{}{arg1, arg2, arg3, arg4Copy})
fake.getBufferedPacketsMutex.Unlock()
if stub != nil {
return stub(arg1, arg2, arg3, arg4)
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeReceiver) GetBufferedPacketsCallCount() int {
fake.getBufferedPacketsMutex.RLock()
defer fake.getBufferedPacketsMutex.RUnlock()
return len(fake.getBufferedPacketsArgsForCall)
}
func (fake *FakeReceiver) GetBufferedPacketsCalls(stub func(uint32, uint16, uint32, []uint16) []rtp.Packet) {
fake.getBufferedPacketsMutex.Lock()
defer fake.getBufferedPacketsMutex.Unlock()
fake.GetBufferedPacketsStub = stub
}
func (fake *FakeReceiver) GetBufferedPacketsArgsForCall(i int) (uint32, uint16, uint32, []uint16) {
fake.getBufferedPacketsMutex.RLock()
defer fake.getBufferedPacketsMutex.RUnlock()
argsForCall := fake.getBufferedPacketsArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4
}
func (fake *FakeReceiver) GetBufferedPacketsReturns(result1 []rtp.Packet) {
fake.getBufferedPacketsMutex.Lock()
defer fake.getBufferedPacketsMutex.Unlock()
fake.GetBufferedPacketsStub = nil
fake.getBufferedPacketsReturns = struct {
result1 []rtp.Packet
}{result1}
}
func (fake *FakeReceiver) GetBufferedPacketsReturnsOnCall(i int, result1 []rtp.Packet) {
fake.getBufferedPacketsMutex.Lock()
defer fake.getBufferedPacketsMutex.Unlock()
fake.GetBufferedPacketsStub = nil
if fake.getBufferedPacketsReturnsOnCall == nil {
fake.getBufferedPacketsReturnsOnCall = make(map[int]struct {
result1 []rtp.Packet
})
}
fake.getBufferedPacketsReturnsOnCall[i] = struct {
result1 []rtp.Packet
}{result1}
}
func (fake *FakeReceiver) ReadRTP() (*rtp.Packet, error) {
fake.readRTPMutex.Lock()
ret, specificReturn := fake.readRTPReturnsOnCall[len(fake.readRTPArgsForCall)]
fake.readRTPArgsForCall = append(fake.readRTPArgsForCall, struct {
}{})
stub := fake.ReadRTPStub
fakeReturns := fake.readRTPReturns
fake.recordInvocation("ReadRTP", []interface{}{})
fake.readRTPMutex.Unlock()
if stub != nil {
return stub()
}
if specificReturn {
return ret.result1, ret.result2
}
return fakeReturns.result1, fakeReturns.result2
}
func (fake *FakeReceiver) ReadRTPCallCount() int {
fake.readRTPMutex.RLock()
defer fake.readRTPMutex.RUnlock()
return len(fake.readRTPArgsForCall)
}
func (fake *FakeReceiver) ReadRTPCalls(stub func() (*rtp.Packet, error)) {
fake.readRTPMutex.Lock()
defer fake.readRTPMutex.Unlock()
fake.ReadRTPStub = stub
}
func (fake *FakeReceiver) ReadRTPReturns(result1 *rtp.Packet, result2 error) {
fake.readRTPMutex.Lock()
defer fake.readRTPMutex.Unlock()
fake.ReadRTPStub = nil
fake.readRTPReturns = struct {
result1 *rtp.Packet
result2 error
}{result1, result2}
}
func (fake *FakeReceiver) ReadRTPReturnsOnCall(i int, result1 *rtp.Packet, result2 error) {
fake.readRTPMutex.Lock()
defer fake.readRTPMutex.Unlock()
fake.ReadRTPStub = nil
if fake.readRTPReturnsOnCall == nil {
fake.readRTPReturnsOnCall = make(map[int]struct {
result1 *rtp.Packet
result2 error
})
}
fake.readRTPReturnsOnCall[i] = struct {
result1 *rtp.Packet
result2 error
}{result1, result2}
}
func (fake *FakeReceiver) Start() {
fake.startMutex.Lock()
fake.startArgsForCall = append(fake.startArgsForCall, struct {
}{})
stub := fake.StartStub
fake.recordInvocation("Start", []interface{}{})
fake.startMutex.Unlock()
if stub != nil {
fake.StartStub()
}
}
func (fake *FakeReceiver) StartCallCount() int {
fake.startMutex.RLock()
defer fake.startMutex.RUnlock()
return len(fake.startArgsForCall)
}
func (fake *FakeReceiver) StartCalls(stub func()) {
fake.startMutex.Lock()
defer fake.startMutex.Unlock()
fake.StartStub = stub
}
func (fake *FakeReceiver) TrackId() string {
fake.trackIdMutex.Lock()
ret, specificReturn := fake.trackIdReturnsOnCall[len(fake.trackIdArgsForCall)]
fake.trackIdArgsForCall = append(fake.trackIdArgsForCall, struct {
}{})
stub := fake.TrackIdStub
fakeReturns := fake.trackIdReturns
fake.recordInvocation("TrackId", []interface{}{})
fake.trackIdMutex.Unlock()
if stub != nil {
return stub()
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeReceiver) TrackIdCallCount() int {
fake.trackIdMutex.RLock()
defer fake.trackIdMutex.RUnlock()
return len(fake.trackIdArgsForCall)
}
func (fake *FakeReceiver) TrackIdCalls(stub func() string) {
fake.trackIdMutex.Lock()
defer fake.trackIdMutex.Unlock()
fake.TrackIdStub = stub
}
func (fake *FakeReceiver) TrackIdReturns(result1 string) {
fake.trackIdMutex.Lock()
defer fake.trackIdMutex.Unlock()
fake.TrackIdStub = nil
fake.trackIdReturns = struct {
result1 string
}{result1}
}
func (fake *FakeReceiver) TrackIdReturnsOnCall(i int, result1 string) {
fake.trackIdMutex.Lock()
defer fake.trackIdMutex.Unlock()
fake.TrackIdStub = nil
if fake.trackIdReturnsOnCall == nil {
fake.trackIdReturnsOnCall = make(map[int]struct {
result1 string
})
}
fake.trackIdReturnsOnCall[i] = struct {
result1 string
}{result1}
}
func (fake *FakeReceiver) Invocations() map[string][][]interface{} {
fake.invocationsMutex.RLock()
defer fake.invocationsMutex.RUnlock()
fake.getBufferedPacketsMutex.RLock()
defer fake.getBufferedPacketsMutex.RUnlock()
fake.readRTPMutex.RLock()
defer fake.readRTPMutex.RUnlock()
fake.startMutex.RLock()
defer fake.startMutex.RUnlock()
fake.trackIdMutex.RLock()
defer fake.trackIdMutex.RUnlock()
copiedInvocations := map[string][][]interface{}{}
for key, value := range fake.invocations {
copiedInvocations[key] = value
}
return copiedInvocations
}
func (fake *FakeReceiver) recordInvocation(key string, args []interface{}) {
fake.invocationsMutex.Lock()
defer fake.invocationsMutex.Unlock()
if fake.invocations == nil {
fake.invocations = map[string][][]interface{}{}
}
if fake.invocations[key] == nil {
fake.invocations[key] = [][]interface{}{}
}
fake.invocations[key] = append(fake.invocations[key], args)
}
var _ types.Receiver = new(FakeReceiver)
pkg/rtc/rtcfakes/fake_signal_connection.go → pkg/rtc/types/typesfakes/fake_signal_connection.go
+3 -3
View File
@@ -1,10 +1,10 @@
// Code generated by counterfeiter. DO NOT EDIT.
package rtcfakes
package typesfakes
import (
"sync"
"github.com/livekit/livekit-server/pkg/rtc"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/proto/livekit"
)
@@ -179,4 +179,4 @@ func (fake *FakeSignalConnection) recordInvocation(key string, args []interface{
fake.invocations[key] = append(fake.invocations[key], args)
}
var _ rtc.SignalConnection = new(FakeSignalConnection)
var _ types.SignalConnection = new(FakeSignalConnection)
pkg/rtc/rtcfakes/fake_websocket_client.go → pkg/rtc/types/typesfakes/fake_websocket_client.go
+3 -3
View File
@@ -1,11 +1,11 @@
// Code generated by counterfeiter. DO NOT EDIT.
package rtcfakes
package typesfakes
import (
"sync"
"time"
"github.com/livekit/livekit-server/pkg/rtc"
"github.com/livekit/livekit-server/pkg/rtc/types"
)
type FakeWebsocketClient struct {
@@ -274,4 +274,4 @@ func (fake *FakeWebsocketClient) recordInvocation(key string, args []interface{}
fake.invocations[key] = append(fake.invocations[key], args)
}
var _ rtc.WebsocketClient = new(FakeWebsocketClient)
var _ types.WebsocketClient = new(FakeWebsocketClient)
pkg/rtc/utils.go
+13 -2
View File
@@ -7,6 +7,7 @@ import (
"github.com/pion/webrtc/v3"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/proto/livekit"
)
@@ -41,7 +42,7 @@ func UnpackDataTrackLabel(packed string) (peerId string, trackId string, label s
return
}
func ToProtoParticipants(participants []Participant) []*livekit.ParticipantInfo {
func ToProtoParticipants(participants []types.Participant) []*livekit.ParticipantInfo {
infos := make([]*livekit.ParticipantInfo, 0, len(participants))
for _, op := range participants {
infos = append(infos, op.ToProto())
@@ -87,7 +88,7 @@ func FromProtoTrickle(trickle *livekit.Trickle) webrtc.ICECandidateInit {
return ci
}
func ToProtoTrack(t PublishedTrack) *livekit.TrackInfo {
func ToProtoTrack(t types.PublishedTrack) *livekit.TrackInfo {
return &livekit.TrackInfo{
Sid: t.ID(),
Type: t.Kind(),
@@ -96,6 +97,16 @@ func ToProtoTrack(t PublishedTrack) *livekit.TrackInfo {
}
}
func ToProtoTrackKind(kind webrtc.RTPCodecType) livekit.TrackInfo_Type {
switch kind {
case webrtc.RTPCodecTypeVideo:
return livekit.TrackInfo_VIDEO
case webrtc.RTPCodecTypeAudio:
return livekit.TrackInfo_AUDIO
}
panic("unsupported track kind")
}
func IsEOF(err error) bool {
return err == io.ErrClosedPipe || err == io.EOF
}
pkg/rtc/wsprotocol.go
+3 -2
View File
@@ -9,6 +9,7 @@ import (
"google.golang.org/protobuf/proto"
"github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/proto/livekit"
)
@@ -18,12 +19,12 @@ const (
)
type WSSignalConnection struct {
conn WebsocketClient
conn types.WebsocketClient
mu sync.Mutex
useJSON bool
}
func NewWSSignalConnection(conn WebsocketClient) *WSSignalConnection {
func NewWSSignalConnection(conn types.WebsocketClient) *WSSignalConnection {
wsc := &WSSignalConnection{
conn: conn,
mu: sync.Mutex{},
pkg/service/rtc.go
+3 -2
View File
@@ -11,6 +11,7 @@ import (
"github.com/livekit/livekit-server/pkg/config"
"github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/rtc"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/proto/livekit"
)
@@ -165,7 +166,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
}
}
func (s *RTCService) handleOffer(participant rtc.Participant, offer *livekit.SessionDescription) error {
func (s *RTCService) handleOffer(participant types.Participant, offer *livekit.SessionDescription) error {
log := logger.GetLogger()
_, err := participant.Answer(rtc.FromProtoSessionDescription(offer))
@@ -177,7 +178,7 @@ func (s *RTCService) handleOffer(participant rtc.Participant, offer *livekit.Ses
return nil
}
func (s *RTCService) handleTrickle(participant rtc.Participant, trickle *livekit.Trickle) error {
func (s *RTCService) handleTrickle(participant types.Participant, trickle *livekit.Trickle) error {
candidateInit := rtc.FromProtoTrickle(trickle)
logger.GetLogger().Debugw("adding peer candidate", "participant", participant.ID())
if err := participant.AddICECandidate(candidateInit); err != nil {