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

participant key (#106)

Browse Source 
* participant key

* update fake router

* WriteRTCMessage

* move sink closing

* unused code
This commit is contained in:
David Colburn
2021-08-31 16:14:06 -05:00
committed by GitHub
parent eb6e763e8d
commit 4044c314b8
7 changed files with 101 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/routing/interfaces.go
+2 -2
View File
@@ -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)
pkg/routing/localrouter.go
+9 -3
View File
@@ -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) {
pkg/routing/node.go
-8
View File
@@ -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) {
pkg/routing/redisrouter.go
+6 -9
View File
@@ -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 {
pkg/routing/routingfakes/fake_router.go
+80 -83
View File
@@ -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
pkg/routing/utils.go
+1 -1
View File
@@ -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
}
pkg/service/roomservice.go
+3 -15
View File
@@ -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)
}