From 4044c314b8a1fdf0ad6722249f68baf68f938139 Mon Sep 17 00:00:00 2001 From: David Colburn Date: Tue, 31 Aug 2021 16:14:06 -0500 Subject: [PATCH] participant key (#106) * participant key * update fake router * WriteRTCMessage * move sink closing * unused code --- pkg/routing/interfaces.go | 4 +- pkg/routing/localrouter.go | 12 +- pkg/routing/node.go | 8 -- pkg/routing/redisrouter.go | 15 +-- pkg/routing/routingfakes/fake_router.go | 163 ++++++++++++------------ pkg/routing/utils.go | 2 +- pkg/service/roomservice.go | 18 +-- 7 files changed, 101 insertions(+), 121 deletions(-) diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index 1f32b050c..c94c9d94a 100644 --- a/pkg/routing/interfaces.go +++ b/pkg/routing/interfaces.go @@ -53,8 +53,8 @@ type Router interface { // StartParticipantSignal participant signal connection is ready to start StartParticipantSignal(ctx context.Context, roomName string, pi ParticipantInit) (connectionId string, reqSink MessageSink, resSource MessageSource, err error) - // CreateRTCSink sends a message to RTC node - CreateRTCSink(ctx context.Context, roomName, identity string) (MessageSink, error) + // WriteRTCMessage sends a message to the RTC node + WriteRTCMessage(ctx context.Context, roomName, identity string, msg *livekit.RTCNodeMessage) error // OnNewParticipantRTC is called to start a new participant's RTC connection OnNewParticipantRTC(callback NewParticipantCallback) diff --git a/pkg/routing/localrouter.go b/pkg/routing/localrouter.go index bdd18f9d3..93639b55c 100644 --- a/pkg/routing/localrouter.go +++ b/pkg/routing/localrouter.go @@ -85,7 +85,7 @@ func (r *LocalRouter) StartParticipantSignal(ctx context.Context, roomName strin } // index channels by roomName | identity - key := ParticipantKey(roomName, pi.Identity) + key := participantKey(roomName, pi.Identity) reqChan := r.getOrCreateMessageChannel(r.requestChannels, key) resChan := r.getOrCreateMessageChannel(r.responseChannels, key) @@ -101,12 +101,18 @@ func (r *LocalRouter) StartParticipantSignal(ctx context.Context, roomName strin return pi.Identity, reqChan, resChan, nil } -func (r *LocalRouter) CreateRTCSink(ctx context.Context, roomName, identity string) (MessageSink, error) { +func (r *LocalRouter) WriteRTCMessage(ctx context.Context, roomName, identity string, msg *livekit.RTCNodeMessage) error { if r.rtcMessageChan.isClosed.Get() { // create a new one r.rtcMessageChan = NewMessageChannel() } - return r.rtcMessageChan, nil + return r.writeRTCMessage(roomName, identity, msg, r.rtcMessageChan) +} + +func (r *LocalRouter) writeRTCMessage(roomName, identity string, msg *livekit.RTCNodeMessage, sink MessageSink) error { + defer sink.Close() + msg.ParticipantKey = participantKey(roomName, identity) + return sink.WriteMessage(msg) } func (r *LocalRouter) OnNewParticipantRTC(callback NewParticipantCallback) { diff --git a/pkg/routing/node.go b/pkg/routing/node.go index 3fb299bbc..4681e8604 100644 --- a/pkg/routing/node.go +++ b/pkg/routing/node.go @@ -14,14 +14,6 @@ import ( "github.com/livekit/livekit-server/pkg/config" ) -type NodeStats struct { - NumRooms int32 - NumClients int32 - NumVideoChannels int32 - NumAudioChannels int32 - BytesPerMin int64 -} - type LocalNode *livekit.Node func NewLocalNode(conf *config.Config) (LocalNode, error) { diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index d1c78f980..a212caee7 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -29,9 +29,6 @@ type RedisRouter struct { ctx context.Context isStarted utils.AtomicFlag - // map of connectionId => SignalNodeSink - signalSinks map[string]*SignalNodeSink - pubsub *redis.PubSub cancel func() } @@ -40,7 +37,6 @@ func NewRedisRouter(currentNode LocalNode, rc *redis.Client) *RedisRouter { rr := &RedisRouter{ LocalRouter: *NewLocalRouter(currentNode), rc: rc, - signalSinks: make(map[string]*SignalNodeSink), } rr.ctx, rr.cancel = context.WithCancel(context.Background()) return rr @@ -139,7 +135,7 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName strin // create a new connection id connectionId = utils.NewGuid("CO_") - pKey := ParticipantKey(roomName, pi.Identity) + pKey := participantKey(roomName, pi.Identity) // map signal & rtc nodes if err = r.setParticipantSignalNode(connectionId, r.currentNode.Id); err != nil { @@ -171,14 +167,15 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName strin return connectionId, sink, resChan, nil } -func (r *RedisRouter) CreateRTCSink(ctx context.Context, roomName, identity string) (MessageSink, error) { - pkey := ParticipantKey(roomName, identity) +func (r *RedisRouter) WriteRTCMessage(ctx context.Context, roomName, identity string, msg *livekit.RTCNodeMessage) error { + pkey := participantKey(roomName, identity) rtcNode, err := r.getParticipantRTCNode(pkey) if err != nil { - return nil, err + return err } - return NewRTCNodeSink(r.rc, rtcNode, pkey), nil + rtcSink := NewRTCNodeSink(r.rc, rtcNode, pkey) + return r.writeRTCMessage(roomName, identity, msg, rtcSink) } func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantKey string) error { diff --git a/pkg/routing/routingfakes/fake_router.go b/pkg/routing/routingfakes/fake_router.go index d43cea8e8..80ff6dc70 100644 --- a/pkg/routing/routingfakes/fake_router.go +++ b/pkg/routing/routingfakes/fake_router.go @@ -22,21 +22,6 @@ type FakeRouter struct { clearRoomStateReturnsOnCall map[int]struct { result1 error } - CreateRTCSinkStub func(context.Context, string, string) (routing.MessageSink, error) - createRTCSinkMutex sync.RWMutex - createRTCSinkArgsForCall []struct { - arg1 context.Context - arg2 string - arg3 string - } - createRTCSinkReturns struct { - result1 routing.MessageSink - result2 error - } - createRTCSinkReturnsOnCall map[int]struct { - result1 routing.MessageSink - result2 error - } GetNodeStub func(string) (*livekit.Node, error) getNodeMutex sync.RWMutex getNodeArgsForCall []struct { @@ -162,6 +147,20 @@ type FakeRouter struct { unregisterNodeReturnsOnCall map[int]struct { result1 error } + WriteRTCMessageStub func(context.Context, string, string, *livekit.RTCNodeMessage) error + writeRTCMessageMutex sync.RWMutex + writeRTCMessageArgsForCall []struct { + arg1 context.Context + arg2 string + arg3 string + arg4 *livekit.RTCNodeMessage + } + writeRTCMessageReturns struct { + result1 error + } + writeRTCMessageReturnsOnCall map[int]struct { + result1 error + } invocations map[string][][]interface{} invocationsMutex sync.RWMutex } @@ -228,72 +227,6 @@ func (fake *FakeRouter) ClearRoomStateReturnsOnCall(i int, result1 error) { }{result1} } -func (fake *FakeRouter) CreateRTCSink(arg1 context.Context, arg2 string, arg3 string) (routing.MessageSink, error) { - fake.createRTCSinkMutex.Lock() - ret, specificReturn := fake.createRTCSinkReturnsOnCall[len(fake.createRTCSinkArgsForCall)] - fake.createRTCSinkArgsForCall = append(fake.createRTCSinkArgsForCall, struct { - arg1 context.Context - arg2 string - arg3 string - }{arg1, arg2, arg3}) - stub := fake.CreateRTCSinkStub - fakeReturns := fake.createRTCSinkReturns - fake.recordInvocation("CreateRTCSink", []interface{}{arg1, arg2, arg3}) - fake.createRTCSinkMutex.Unlock() - if stub != nil { - return stub(arg1, arg2, arg3) - } - if specificReturn { - return ret.result1, ret.result2 - } - return fakeReturns.result1, fakeReturns.result2 -} - -func (fake *FakeRouter) CreateRTCSinkCallCount() int { - fake.createRTCSinkMutex.RLock() - defer fake.createRTCSinkMutex.RUnlock() - return len(fake.createRTCSinkArgsForCall) -} - -func (fake *FakeRouter) CreateRTCSinkCalls(stub func(context.Context, string, string) (routing.MessageSink, error)) { - fake.createRTCSinkMutex.Lock() - defer fake.createRTCSinkMutex.Unlock() - fake.CreateRTCSinkStub = stub -} - -func (fake *FakeRouter) CreateRTCSinkArgsForCall(i int) (context.Context, string, string) { - fake.createRTCSinkMutex.RLock() - defer fake.createRTCSinkMutex.RUnlock() - argsForCall := fake.createRTCSinkArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 -} - -func (fake *FakeRouter) CreateRTCSinkReturns(result1 routing.MessageSink, result2 error) { - fake.createRTCSinkMutex.Lock() - defer fake.createRTCSinkMutex.Unlock() - fake.CreateRTCSinkStub = nil - fake.createRTCSinkReturns = struct { - result1 routing.MessageSink - result2 error - }{result1, result2} -} - -func (fake *FakeRouter) CreateRTCSinkReturnsOnCall(i int, result1 routing.MessageSink, result2 error) { - fake.createRTCSinkMutex.Lock() - defer fake.createRTCSinkMutex.Unlock() - fake.CreateRTCSinkStub = nil - if fake.createRTCSinkReturnsOnCall == nil { - fake.createRTCSinkReturnsOnCall = make(map[int]struct { - result1 routing.MessageSink - result2 error - }) - } - fake.createRTCSinkReturnsOnCall[i] = struct { - result1 routing.MessageSink - result2 error - }{result1, result2} -} - func (fake *FakeRouter) GetNode(arg1 string) (*livekit.Node, error) { fake.getNodeMutex.Lock() ret, specificReturn := fake.getNodeReturnsOnCall[len(fake.getNodeArgsForCall)] @@ -914,13 +847,75 @@ func (fake *FakeRouter) UnregisterNodeReturnsOnCall(i int, result1 error) { }{result1} } +func (fake *FakeRouter) WriteRTCMessage(arg1 context.Context, arg2 string, arg3 string, arg4 *livekit.RTCNodeMessage) error { + fake.writeRTCMessageMutex.Lock() + ret, specificReturn := fake.writeRTCMessageReturnsOnCall[len(fake.writeRTCMessageArgsForCall)] + fake.writeRTCMessageArgsForCall = append(fake.writeRTCMessageArgsForCall, struct { + arg1 context.Context + arg2 string + arg3 string + arg4 *livekit.RTCNodeMessage + }{arg1, arg2, arg3, arg4}) + stub := fake.WriteRTCMessageStub + fakeReturns := fake.writeRTCMessageReturns + fake.recordInvocation("WriteRTCMessage", []interface{}{arg1, arg2, arg3, arg4}) + fake.writeRTCMessageMutex.Unlock() + if stub != nil { + return stub(arg1, arg2, arg3, arg4) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeRouter) WriteRTCMessageCallCount() int { + fake.writeRTCMessageMutex.RLock() + defer fake.writeRTCMessageMutex.RUnlock() + return len(fake.writeRTCMessageArgsForCall) +} + +func (fake *FakeRouter) WriteRTCMessageCalls(stub func(context.Context, string, string, *livekit.RTCNodeMessage) error) { + fake.writeRTCMessageMutex.Lock() + defer fake.writeRTCMessageMutex.Unlock() + fake.WriteRTCMessageStub = stub +} + +func (fake *FakeRouter) WriteRTCMessageArgsForCall(i int) (context.Context, string, string, *livekit.RTCNodeMessage) { + fake.writeRTCMessageMutex.RLock() + defer fake.writeRTCMessageMutex.RUnlock() + argsForCall := fake.writeRTCMessageArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 +} + +func (fake *FakeRouter) WriteRTCMessageReturns(result1 error) { + fake.writeRTCMessageMutex.Lock() + defer fake.writeRTCMessageMutex.Unlock() + fake.WriteRTCMessageStub = nil + fake.writeRTCMessageReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeRouter) WriteRTCMessageReturnsOnCall(i int, result1 error) { + fake.writeRTCMessageMutex.Lock() + defer fake.writeRTCMessageMutex.Unlock() + fake.WriteRTCMessageStub = nil + if fake.writeRTCMessageReturnsOnCall == nil { + fake.writeRTCMessageReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.writeRTCMessageReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeRouter) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() fake.clearRoomStateMutex.RLock() defer fake.clearRoomStateMutex.RUnlock() - fake.createRTCSinkMutex.RLock() - defer fake.createRTCSinkMutex.RUnlock() fake.getNodeMutex.RLock() defer fake.getNodeMutex.RUnlock() fake.getNodeForRoomMutex.RLock() @@ -945,6 +940,8 @@ func (fake *FakeRouter) Invocations() map[string][][]interface{} { defer fake.stopMutex.RUnlock() fake.unregisterNodeMutex.RLock() defer fake.unregisterNodeMutex.RUnlock() + fake.writeRTCMessageMutex.RLock() + defer fake.writeRTCMessageMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} for key, value := range fake.invocations { copiedInvocations[key] = value diff --git a/pkg/routing/utils.go b/pkg/routing/utils.go index 7ba17bbee..da47f9e15 100644 --- a/pkg/routing/utils.go +++ b/pkg/routing/utils.go @@ -22,7 +22,7 @@ func GetAvailableNodes(nodes []*livekit.Node) []*livekit.Node { }).([]*livekit.Node) } -func ParticipantKey(roomName, identity string) string { +func participantKey(roomName, identity string) string { return roomName + "|" + identity } diff --git a/pkg/service/roomservice.go b/pkg/service/roomservice.go index 70e134f54..24d846f10 100644 --- a/pkg/service/roomservice.go +++ b/pkg/service/roomservice.go @@ -149,7 +149,6 @@ func (s *RoomService) MutePublishedTrack(ctx context.Context, req *livekit.MuteR } err = s.writeMessage(ctx, req.Room, req.Identity, &livekit.RTCNodeMessage{ - ParticipantKey: routing.ParticipantKey(req.Room, req.Identity), Message: &livekit.RTCNodeMessage_MuteTrack{ MuteTrack: req, }, @@ -219,26 +218,15 @@ func (s *RoomService) SendData(ctx context.Context, req *livekit.SendDataRequest return &livekit.SendDataResponse{}, nil } -func (s *RoomService) createRTCSink(ctx context.Context, room, identity string) (routing.MessageSink, error) { +func (s *RoomService) writeMessage(ctx context.Context, room, identity string, msg *livekit.RTCNodeMessage) error { if err := EnsureAdminPermission(ctx, room); err != nil { - return nil, twirpAuthError(err) + return twirpAuthError(err) } _, err := s.roomManager.LoadParticipant(ctx, room, identity) - if err != nil { - return nil, err - } - - return s.router.CreateRTCSink(ctx, room, identity) -} - -func (s *RoomService) writeMessage(ctx context.Context, room, identity string, msg *livekit.RTCNodeMessage) error { - rtcSink, err := s.createRTCSink(ctx, room, identity) if err != nil { return err } - defer rtcSink.Close() - msg.ParticipantKey = routing.ParticipantKey(room, identity) - return rtcSink.WriteMessage(msg) + return s.router.WriteRTCMessage(ctx, room, identity, msg) }