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

Export CloseSignalConnection (#936)

Browse Source 
* Export CloseSignalConnection

There are a few places where that close pattern is repeated.
Export it and use that function in other places directly.

* fix test
This commit is contained in:
Raja Subramanian
2022-08-21 11:33:35 +05:30
committed by GitHub
parent b8bda3f14b
commit 70422c0267
6 changed files with 41 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/participant.go
+3 -3
View File
@@ -612,7 +612,7 @@ func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseRea
p.updateState(livekit.ParticipantInfo_DISCONNECTED)
// ensure this is synchronized
p.closeSignalConnection()
p.CloseSignalConnection()
p.lock.RLock()
onClose := p.onClose
p.lock.RUnlock()
@@ -1206,7 +1206,7 @@ func (p *ParticipantImpl) setupDisconnectTimer() {
func (p *ParticipantImpl) onAnyTransportFailed() {
// clients support resuming of connections when websocket becomes disconnected
p.closeSignalConnection()
p.CloseSignalConnection()
// detect when participant has actually left.
p.setupDisconnectTimer()
@@ -1891,7 +1891,7 @@ func (p *ParticipantImpl) handleSubscriberNegotiationFailed() {
},
},
})
p.closeSignalConnection()
p.CloseSignalConnection()
}
func (p *ParticipantImpl) EnqueueSubscribeTrack(trackID livekit.TrackID, f func(sub types.LocalParticipant) error) {
pkg/rtc/participant_internal_test.go
+1 -1
View File
@@ -184,7 +184,7 @@ func TestTrackPublishing(t *testing.T) {
func TestOutOfOrderUpdates(t *testing.T) {
p := newParticipantForTest("test")
p.SetMetadata("initial metadata")
sink := p.GetResponseSink().(*routingfakes.FakeMessageSink)
sink := p.getResponseSink().(*routingfakes.FakeMessageSink)
pi1 := p.ToProto()
p.SetMetadata("second update")
pi2 := p.ToProto()
pkg/rtc/participant_signal.go
+5 -4
View File
@@ -10,7 +10,7 @@ import (
"github.com/livekit/livekit-server/pkg/routing"
)
func (p *ParticipantImpl) GetResponseSink() routing.MessageSink {
func (p *ParticipantImpl) getResponseSink() routing.MessageSink {
if !p.resSinkValid.Load() {
return nil
}
@@ -174,7 +174,7 @@ func (p *ParticipantImpl) writeMessage(msg *livekit.SignalResponse) error {
if p.State() == livekit.ParticipantInfo_DISCONNECTED {
return nil
}
sink := p.GetResponseSink()
sink := p.getResponseSink()
if sink == nil {
return nil
}
@@ -188,9 +188,10 @@ func (p *ParticipantImpl) writeMessage(msg *livekit.SignalResponse) error {
}
// closes signal connection to notify client to resume/reconnect
func (p *ParticipantImpl) closeSignalConnection() {
sink := p.GetResponseSink()
func (p *ParticipantImpl) CloseSignalConnection() {
sink := p.getResponseSink()
if sink != nil {
p.params.Logger.Infow("closing signal connection")
sink.Close()
p.SetResponseSink(nil)
}
pkg/rtc/room.go
+1 -3
View File
@@ -333,9 +333,7 @@ func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions
func (r *Room) ResumeParticipant(p types.LocalParticipant, responseSink routing.MessageSink) error {
// close previous sink, and link to new one
if prevSink := p.GetResponseSink(); prevSink != nil {
prevSink.Close()
}
p.CloseSignalConnection()
p.SetResponseSink(responseSink)
updates := ToProtoParticipants(r.GetParticipants())
pkg/rtc/types/interfaces.go
+1 -1
View File
@@ -214,8 +214,8 @@ type LocalParticipant interface {
SubscriberAsPrimary() bool
GetClientConfiguration() *livekit.ClientConfiguration
GetResponseSink() routing.MessageSink
SetResponseSink(sink routing.MessageSink)
CloseSignalConnection()
// permissions
ClaimGrants() *auth.ClaimGrants
pkg/rtc/types/typesfakes/fake_local_participant.go
+30 -65
View File
@@ -150,6 +150,10 @@ type FakeLocalParticipant struct {
closeReturnsOnCall map[int]struct {
result1 error
}
CloseSignalConnectionStub func()
closeSignalConnectionMutex sync.RWMutex
closeSignalConnectionArgsForCall []struct {
}
ConnectedAtStub func() time.Time
connectedAtMutex sync.RWMutex
connectedAtArgsForCall []struct {
@@ -269,16 +273,6 @@ type FakeLocalParticipant struct {
getPublishedTracksReturnsOnCall map[int]struct {
result1 []types.MediaTrack
}
GetResponseSinkStub func() routing.MessageSink
getResponseSinkMutex sync.RWMutex
getResponseSinkArgsForCall []struct {
}
getResponseSinkReturns struct {
result1 routing.MessageSink
}
getResponseSinkReturnsOnCall map[int]struct {
result1 routing.MessageSink
}
GetSubscribedParticipantsStub func() []livekit.ParticipantID
getSubscribedParticipantsMutex sync.RWMutex
getSubscribedParticipantsArgsForCall []struct {
@@ -1461,6 +1455,30 @@ func (fake *FakeLocalParticipant) CloseReturnsOnCall(i int, result1 error) {
}{result1}
}
func (fake *FakeLocalParticipant) CloseSignalConnection() {
fake.closeSignalConnectionMutex.Lock()
fake.closeSignalConnectionArgsForCall = append(fake.closeSignalConnectionArgsForCall, struct {
}{})
stub := fake.CloseSignalConnectionStub
fake.recordInvocation("CloseSignalConnection", []interface{}{})
fake.closeSignalConnectionMutex.Unlock()
if stub != nil {
fake.CloseSignalConnectionStub()
}
}
func (fake *FakeLocalParticipant) CloseSignalConnectionCallCount() int {
fake.closeSignalConnectionMutex.RLock()
defer fake.closeSignalConnectionMutex.RUnlock()
return len(fake.closeSignalConnectionArgsForCall)
}
func (fake *FakeLocalParticipant) CloseSignalConnectionCalls(stub func()) {
fake.closeSignalConnectionMutex.Lock()
defer fake.closeSignalConnectionMutex.Unlock()
fake.CloseSignalConnectionStub = stub
}
func (fake *FakeLocalParticipant) ConnectedAt() time.Time {
fake.connectedAtMutex.Lock()
ret, specificReturn := fake.connectedAtReturnsOnCall[len(fake.connectedAtArgsForCall)]
@@ -2080,59 +2098,6 @@ func (fake *FakeLocalParticipant) GetPublishedTracksReturnsOnCall(i int, result1
}{result1}
}
func (fake *FakeLocalParticipant) GetResponseSink() routing.MessageSink {
fake.getResponseSinkMutex.Lock()
ret, specificReturn := fake.getResponseSinkReturnsOnCall[len(fake.getResponseSinkArgsForCall)]
fake.getResponseSinkArgsForCall = append(fake.getResponseSinkArgsForCall, struct {
}{})
stub := fake.GetResponseSinkStub
fakeReturns := fake.getResponseSinkReturns
fake.recordInvocation("GetResponseSink", []interface{}{})
fake.getResponseSinkMutex.Unlock()
if stub != nil {
return stub()
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeLocalParticipant) GetResponseSinkCallCount() int {
fake.getResponseSinkMutex.RLock()
defer fake.getResponseSinkMutex.RUnlock()
return len(fake.getResponseSinkArgsForCall)
}
func (fake *FakeLocalParticipant) GetResponseSinkCalls(stub func() routing.MessageSink) {
fake.getResponseSinkMutex.Lock()
defer fake.getResponseSinkMutex.Unlock()
fake.GetResponseSinkStub = stub
}
func (fake *FakeLocalParticipant) GetResponseSinkReturns(result1 routing.MessageSink) {
fake.getResponseSinkMutex.Lock()
defer fake.getResponseSinkMutex.Unlock()
fake.GetResponseSinkStub = nil
fake.getResponseSinkReturns = struct {
result1 routing.MessageSink
}{result1}
}
func (fake *FakeLocalParticipant) GetResponseSinkReturnsOnCall(i int, result1 routing.MessageSink) {
fake.getResponseSinkMutex.Lock()
defer fake.getResponseSinkMutex.Unlock()
fake.GetResponseSinkStub = nil
if fake.getResponseSinkReturnsOnCall == nil {
fake.getResponseSinkReturnsOnCall = make(map[int]struct {
result1 routing.MessageSink
})
}
fake.getResponseSinkReturnsOnCall[i] = struct {
result1 routing.MessageSink
}{result1}
}
func (fake *FakeLocalParticipant) GetSubscribedParticipants() []livekit.ParticipantID {
fake.getSubscribedParticipantsMutex.Lock()
ret, specificReturn := fake.getSubscribedParticipantsReturnsOnCall[len(fake.getSubscribedParticipantsArgsForCall)]
@@ -4837,6 +4802,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
defer fake.clearInProgressAndProcessSubscriptionRequestsQueueMutex.RUnlock()
fake.closeMutex.RLock()
defer fake.closeMutex.RUnlock()
fake.closeSignalConnectionMutex.RLock()
defer fake.closeSignalConnectionMutex.RUnlock()
fake.connectedAtMutex.RLock()
defer fake.connectedAtMutex.RUnlock()
fake.debugInfoMutex.RLock()
@@ -4861,8 +4828,6 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
defer fake.getPublishedTrackMutex.RUnlock()
fake.getPublishedTracksMutex.RLock()
defer fake.getPublishedTracksMutex.RUnlock()
fake.getResponseSinkMutex.RLock()
defer fake.getResponseSinkMutex.RUnlock()
fake.getSubscribedParticipantsMutex.RLock()
defer fake.getSubscribedParticipantsMutex.RUnlock()
fake.getSubscribedTracksMutex.RLock()