diff --git a/go.mod b/go.mod index 826fd9065..8c9772593 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index cb4a3061f..f145f9293 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index a0ee6a131..f3530a6d2 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -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(), diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 10d06f793..8a7f6a661 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -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 } diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index d62e5dbc0..ffd62a356 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -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) diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 564570c9a..f76dd894e 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -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 diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index f68acba81..e25dba25d 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -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) { diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index e053f2f2e..332102582 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -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