dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-29 09:49:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Pass back serverRegion in when participant's joining (#479)

Browse Source 
* Pass back serverRegion in when participant's joining

* fix tests

* protocol update
This commit is contained in:
David Zhao
2022-03-02 13:36:19 -08:00
committed by GitHub
parent c743d2ab3a
commit 7a5b5dbc69
8 changed files with 29 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -14,7 +14,7 @@ require (
github.com/google/wire v0.5.0
github.com/gorilla/websocket v1.4.2
github.com/hashicorp/golang-lru v0.5.4
github.com/livekit/protocol v0.11.14-0.20220228184153-f3ffabe24080
github.com/livekit/protocol v0.11.14-0.20220302192533-dbd455d2c1de
github.com/magefile/mage v1.11.0
github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0
github.com/mitchellh/go-homedir v1.1.0
go.sum
+2 -2
View File
@@ -132,8 +132,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY/mKrcvA=
github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts=
github.com/livekit/protocol v0.11.14-0.20220228184153-f3ffabe24080 h1:h/S/EQYCsuQFWlyWsoXmdLvaCDlneFKbIc1UVyLcTnM=
github.com/livekit/protocol v0.11.14-0.20220228184153-f3ffabe24080/go.mod h1:3pHsWUtQmWaH8mG0cXrQWpbf3Vo+kj0U+In77CEXu90=
github.com/livekit/protocol v0.11.14-0.20220302192533-dbd455d2c1de h1:uyUDLn1HcyxMwbAPX9SVQHfn6O/Zncx7vPH/YXC39wE=
github.com/livekit/protocol v0.11.14-0.20220302192533-dbd455d2c1de/go.mod h1:3pHsWUtQmWaH8mG0cXrQWpbf3Vo+kj0U+In77CEXu90=
github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls=
github.com/magefile/mage v1.11.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A=
github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0=
pkg/rtc/participant.go
+2
View File
@@ -599,6 +599,7 @@ func (p *ParticipantImpl) SendJoinResponse(
roomInfo *livekit.Room,
otherParticipants []*livekit.ParticipantInfo,
iceServers []*livekit.ICEServer,
region string,
) error {
// send Join response
return p.writeMessage(&livekit.SignalResponse{
@@ -608,6 +609,7 @@ func (p *ParticipantImpl) SendJoinResponse(
Participant: p.ToProto(),
OtherParticipants: otherParticipants,
ServerVersion: version.Version,
ServerRegion: region,
IceServers: iceServers,
// indicates both server and client support subscriber as primary
SubscriberPrimary: p.SubscriberAsPrimary(),
pkg/rtc/room.go
+2 -2
View File
@@ -167,7 +167,7 @@ func (r *Room) Release() {
r.holds.Dec()
}
func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions, iceServers []*livekit.ICEServer) error {
func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions, iceServers []*livekit.ICEServer, region string) error {
r.lock.Lock()
defer r.lock.Unlock()
@@ -259,7 +259,7 @@ func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions
}
})
if err := participant.SendJoinResponse(r.Room, otherParticipants, iceServers); err != nil {
if err := participant.SendJoinResponse(r.Room, otherParticipants, iceServers, region); err != nil {
prometheus.ServiceOperationCounter.WithLabelValues("participant_join", "error", "send_response").Add(1)
return err
}
pkg/rtc/room_test.go
+10 -9
View File
@@ -64,10 +64,10 @@ func TestRoomJoin(t *testing.T) {
rm := newRoomWithParticipants(t, testRoomOpts{num: numParticipants})
pNew := newMockParticipant("new", types.DefaultProtocol, false)
rm.Join(pNew, nil, iceServersForRoom)
_ = rm.Join(pNew, nil, iceServersForRoom, "test")
// expect new participant to get a JoinReply
info, participants, iceServers := pNew.SendJoinResponseArgsForCall(0)
info, participants, iceServers, _ := pNew.SendJoinResponseArgsForCall(0)
require.Equal(t, info.Sid, rm.Room.Sid)
require.Len(t, participants, numParticipants)
require.Len(t, rm.GetParticipants(), numParticipants+1)
@@ -79,7 +79,7 @@ func TestRoomJoin(t *testing.T) {
rm := newRoomWithParticipants(t, testRoomOpts{num: numExisting})
p := newMockParticipant("new", types.DefaultProtocol, false)
err := rm.Join(p, &rtc.ParticipantOptions{AutoSubscribe: true}, iceServersForRoom)
err := rm.Join(p, &rtc.ParticipantOptions{AutoSubscribe: true}, iceServersForRoom, "")
require.NoError(t, err)
stateChangeCB := p.OnStateChangeArgsForCall(0)
@@ -135,7 +135,7 @@ func TestRoomJoin(t *testing.T) {
rm.Room.MaxParticipants = 1
p := newMockParticipant("second", types.ProtocolVersion(0), false)
err := rm.Join(p, nil, iceServersForRoom)
err := rm.Join(p, nil, iceServersForRoom, "")
require.Equal(t, rtc.ErrMaxParticipantsExceeded, err)
})
}
@@ -216,7 +216,7 @@ func TestRoomClosure(t *testing.T) {
require.Len(t, rm.GetParticipants(), 0)
require.True(t, isClosed)
require.Equal(t, rtc.ErrRoomClosed, rm.Join(p, nil, iceServersForRoom))
require.Equal(t, rtc.ErrRoomClosed, rm.Join(p, nil, iceServersForRoom, ""))
})
t.Run("room does not close before empty timeout", func(t *testing.T) {
@@ -501,21 +501,22 @@ func TestHiddenParticipants(t *testing.T) {
defer rm.Close()
pNew := newMockParticipant("new", types.DefaultProtocol, false)
rm.Join(pNew, nil, iceServersForRoom)
rm.Join(pNew, nil, iceServersForRoom, "testregion")
// expect new participant to get a JoinReply
info, participants, iceServers := pNew.SendJoinResponseArgsForCall(0)
info, participants, iceServers, region := pNew.SendJoinResponseArgsForCall(0)
require.Equal(t, info.Sid, rm.Room.Sid)
require.Len(t, participants, 2)
require.Len(t, rm.GetParticipants(), 4)
require.NotEmpty(t, iceServers)
require.Equal(t, "testregion", region)
})
t.Run("hidden participant subscribes to tracks", func(t *testing.T) {
rm := newRoomWithParticipants(t, testRoomOpts{num: 2, numHidden: 1})
p := newMockParticipant("new", types.DefaultProtocol, false)
err := rm.Join(p, &rtc.ParticipantOptions{AutoSubscribe: true}, iceServersForRoom)
err := rm.Join(p, &rtc.ParticipantOptions{AutoSubscribe: true}, iceServersForRoom, "")
require.NoError(t, err)
stateChangeCB := p.OnStateChangeArgsForCall(0)
@@ -572,7 +573,7 @@ func newRoomWithParticipants(t *testing.T, opts testRoomOpts) *rtc.Room {
for i := 0; i < opts.num+opts.numHidden; i++ {
identity := livekit.ParticipantIdentity(fmt.Sprintf("p%d", i))
participant := newMockParticipant(identity, opts.protocol, i >= opts.num)
err := rm.Join(participant, &rtc.ParticipantOptions{AutoSubscribe: true}, iceServersForRoom)
err := rm.Join(participant, &rtc.ParticipantOptions{AutoSubscribe: true}, iceServersForRoom, "")
participant.StateReturns(livekit.ParticipantInfo_ACTIVE)
participant.IsReadyReturns(true)
require.NoError(t, err)
pkg/rtc/types/interfaces.go
+1 -1
View File
@@ -129,7 +129,7 @@ type LocalParticipant interface {
GetConnectionQuality() *livekit.ConnectionQualityInfo
// server sent messages
SendJoinResponse(info *livekit.Room, otherParticipants []*livekit.ParticipantInfo, iceServers []*livekit.ICEServer) error
SendJoinResponse(info *livekit.Room, otherParticipants []*livekit.ParticipantInfo, iceServers []*livekit.ICEServer, region string) error
SendParticipantUpdate(participants []*livekit.ParticipantInfo) error
SendSpeakerUpdate(speakers []*livekit.SpeakerInfo) error
SendDataPacket(packet *livekit.DataPacket) error
pkg/rtc/types/typesfakes/fake_local_participant.go
+10 -8
View File
@@ -381,12 +381,13 @@ type FakeLocalParticipant struct {
sendDataPacketReturnsOnCall map[int]struct {
result1 error
}
SendJoinResponseStub func(*livekit.Room, []*livekit.ParticipantInfo, []*livekit.ICEServer) error
SendJoinResponseStub func(*livekit.Room, []*livekit.ParticipantInfo, []*livekit.ICEServer, string) error
sendJoinResponseMutex sync.RWMutex
sendJoinResponseArgsForCall []struct {
arg1 *livekit.Room
arg2 []*livekit.ParticipantInfo
arg3 []*livekit.ICEServer
arg4 string
}
sendJoinResponseReturns struct {
result1 error
@@ -2607,7 +2608,7 @@ func (fake *FakeLocalParticipant) SendDataPacketReturnsOnCall(i int, result1 err
}{result1}
}
func (fake *FakeLocalParticipant) SendJoinResponse(arg1 *livekit.Room, arg2 []*livekit.ParticipantInfo, arg3 []*livekit.ICEServer) error {
func (fake *FakeLocalParticipant) SendJoinResponse(arg1 *livekit.Room, arg2 []*livekit.ParticipantInfo, arg3 []*livekit.ICEServer, arg4 string) error {
var arg2Copy []*livekit.ParticipantInfo
if arg2 != nil {
arg2Copy = make([]*livekit.ParticipantInfo, len(arg2))
@@ -2624,13 +2625,14 @@ func (fake *FakeLocalParticipant) SendJoinResponse(arg1 *livekit.Room, arg2 []*l
arg1 *livekit.Room
arg2 []*livekit.ParticipantInfo
arg3 []*livekit.ICEServer
}{arg1, arg2Copy, arg3Copy})
arg4 string
}{arg1, arg2Copy, arg3Copy, arg4})
stub := fake.SendJoinResponseStub
fakeReturns := fake.sendJoinResponseReturns
fake.recordInvocation("SendJoinResponse", []interface{}{arg1, arg2Copy, arg3Copy})
fake.recordInvocation("SendJoinResponse", []interface{}{arg1, arg2Copy, arg3Copy, arg4})
fake.sendJoinResponseMutex.Unlock()
if stub != nil {
return stub(arg1, arg2, arg3)
return stub(arg1, arg2, arg3, arg4)
}
if specificReturn {
return ret.result1
@@ -2644,17 +2646,17 @@ func (fake *FakeLocalParticipant) SendJoinResponseCallCount() int {
return len(fake.sendJoinResponseArgsForCall)
}
func (fake *FakeLocalParticipant) SendJoinResponseCalls(stub func(*livekit.Room, []*livekit.ParticipantInfo, []*livekit.ICEServer) error) {
func (fake *FakeLocalParticipant) SendJoinResponseCalls(stub func(*livekit.Room, []*livekit.ParticipantInfo, []*livekit.ICEServer, string) error) {
fake.sendJoinResponseMutex.Lock()
defer fake.sendJoinResponseMutex.Unlock()
fake.SendJoinResponseStub = stub
}
func (fake *FakeLocalParticipant) SendJoinResponseArgsForCall(i int) (*livekit.Room, []*livekit.ParticipantInfo, []*livekit.ICEServer) {
func (fake *FakeLocalParticipant) SendJoinResponseArgsForCall(i int) (*livekit.Room, []*livekit.ParticipantInfo, []*livekit.ICEServer, string) {
fake.sendJoinResponseMutex.RLock()
defer fake.sendJoinResponseMutex.RUnlock()
argsForCall := fake.sendJoinResponseArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4
}
func (fake *FakeLocalParticipant) SendJoinResponseReturns(result1 error) {
pkg/service/roommanager.go
+1 -1
View File
@@ -258,7 +258,7 @@ func (r *RoomManager) StartSession(ctx context.Context, roomName livekit.RoomNam
opts := rtc.ParticipantOptions{
AutoSubscribe: pi.AutoSubscribe,
}
if err = room.Join(participant, &opts, r.iceServersForRoom(room.Room)); err != nil {
if err = room.Join(participant, &opts, r.iceServersForRoom(room.Room), r.currentNode.Region); err != nil {
pLogger.Errorw("could not join room", err)
_ = participant.Close(true)
return