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

Types for ConnectionID, NodeID, ParticipantKey (#450)

Browse Source 
* Types for ConnectionID, NodeID, ParticipantKey

* latest protocol version
This commit is contained in:
Raja Subramanian
2022-02-21 08:52:11 +05:30
committed by GitHub
parent f209e9319a
commit 3f38172139
17 changed files with 161 additions and 161 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.13-0.20220208070440-7f8399010124
github.com/livekit/protocol v0.11.14-0.20220221030410-28685be0639d
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.13-0.20220208070440-7f8399010124 h1:nZor+yFxIBKLdQGwwx6GvsnjZhmG9gcz9I2K2zdLiVI=
github.com/livekit/protocol v0.11.13-0.20220208070440-7f8399010124/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg=
github.com/livekit/protocol v0.11.14-0.20220221030410-28685be0639d h1:j1r2zNSzRM12ZbQye9B/HpszTmNv+cBYJ2dPmMFqkf0=
github.com/livekit/protocol v0.11.14-0.20220221030410-28685be0639d/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg=
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/routing/interfaces.go
+2 -2
View File
@@ -55,7 +55,7 @@ type Router interface {
ListNodes() ([]*livekit.Node, error)
GetNodeForRoom(ctx context.Context, roomName livekit.RoomName) (*livekit.Node, error)
SetNodeForRoom(ctx context.Context, roomName livekit.RoomName, nodeId string) error
SetNodeForRoom(ctx context.Context, roomName livekit.RoomName, nodeId livekit.NodeID) error
ClearRoomState(ctx context.Context, roomName livekit.RoomName) error
Start() error
@@ -71,7 +71,7 @@ type Router interface {
type MessageRouter interface {
// StartParticipantSignal participant signal connection is ready to start
StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (connectionId string, reqSink MessageSink, resSource MessageSource, err error)
StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (connectionID livekit.ConnectionID, reqSink MessageSink, resSource MessageSource, err error)
// Write a message to a participant or room
WriteParticipantRTC(ctx context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity, msg *livekit.RTCNodeMessage) error
pkg/routing/localrouter.go
+12 -12
View File
@@ -42,7 +42,7 @@ func (r *LocalRouter) GetNodeForRoom(_ context.Context, _ livekit.RoomName) (*li
return node, nil
}
func (r *LocalRouter) SetNodeForRoom(_ context.Context, _ livekit.RoomName, _ string) error {
func (r *LocalRouter) SetNodeForRoom(_ context.Context, _ livekit.RoomName, _ livekit.NodeID) error {
return nil
}
@@ -63,8 +63,8 @@ func (r *LocalRouter) RemoveDeadNodes() error {
return nil
}
func (r *LocalRouter) GetNode(nodeID string) (*livekit.Node, error) {
if nodeID == r.currentNode.Id {
func (r *LocalRouter) GetNode(nodeID livekit.NodeID) (*livekit.Node, error) {
if nodeID == livekit.NodeID(r.currentNode.Id) {
return r.currentNode, nil
}
return nil, ErrNotFound
@@ -76,7 +76,7 @@ func (r *LocalRouter) ListNodes() ([]*livekit.Node, error) {
}, nil
}
func (r *LocalRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (connectionID string, reqSink MessageSink, resSource MessageSource, err error) {
func (r *LocalRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (connectionID livekit.ConnectionID, reqSink MessageSink, resSource MessageSource, err error) {
// treat it as a new participant connecting
if r.onNewParticipant == nil {
err = ErrHandlerNotDefined
@@ -87,16 +87,16 @@ func (r *LocalRouter) StartParticipantSignal(ctx context.Context, roomName livek
key := participantKey(roomName, pi.Identity)
// close older channels if one already exists
reqChan := r.getMessageChannel(r.requestChannels, key)
reqChan := r.getMessageChannel(r.requestChannels, string(key))
if reqChan != nil {
reqChan.Close()
}
resChan := r.getMessageChannel(r.responseChannels, key)
resChan := r.getMessageChannel(r.responseChannels, string(key))
if resChan != nil {
resChan.Close()
}
reqChan = r.getOrCreateMessageChannel(r.requestChannels, key)
resChan = r.getOrCreateMessageChannel(r.responseChannels, key)
reqChan = r.getOrCreateMessageChannel(r.requestChannels, string(key))
resChan = r.getOrCreateMessageChannel(r.responseChannels, string(key))
r.onNewParticipant(
ctx,
@@ -107,7 +107,7 @@ func (r *LocalRouter) StartParticipantSignal(ctx context.Context, roomName livek
// response sink
resChan,
)
return string(pi.Identity), reqChan, resChan, nil
return livekit.ConnectionID(pi.Identity), reqChan, resChan, nil
}
func (r *LocalRouter) WriteParticipantRTC(_ context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity, msg *livekit.RTCNodeMessage) error {
@@ -115,12 +115,12 @@ func (r *LocalRouter) WriteParticipantRTC(_ context.Context, roomName livekit.Ro
// create a new one
r.rtcMessageChan = NewMessageChannel()
}
msg.ParticipantKey = participantKey(roomName, identity)
msg.ParticipantKey = string(participantKey(roomName, identity))
return r.writeRTCMessage(r.rtcMessageChan, msg)
}
func (r *LocalRouter) WriteRoomRTC(ctx context.Context, roomName livekit.RoomName, msg *livekit.RTCNodeMessage) error {
msg.ParticipantKey = participantKey(roomName, "")
msg.ParticipantKey = string(participantKey(roomName, ""))
return r.WriteNodeRTC(ctx, r.currentNode.Id, msg)
}
@@ -196,7 +196,7 @@ func (r *LocalRouter) rtcMessageWorker() {
// consume messages from
for msg := range r.rtcMessageChan.ReadChan() {
if rtcMsg, ok := msg.(*livekit.RTCNodeMessage); ok {
room, identity, err := parseParticipantKey(rtcMsg.ParticipantKey)
room, identity, err := parseParticipantKey(livekit.ParticipantKey(rtcMsg.ParticipantKey))
if err != nil {
logger.Errorw("could not process RTC message", err)
continue
pkg/routing/redis.go
+29 -29
View File
@@ -20,26 +20,26 @@ const (
var redisCtx = context.Background()
// location of the participant's RTC connection, hash
func participantRTCKey(participantKey string) string {
return "participant_rtc:" + participantKey
func participantRTCKey(participantKey livekit.ParticipantKey) string {
return "participant_rtc:" + string(participantKey)
}
// location of the participant's Signal connection, hash
func participantSignalKey(connectionId string) string {
return "participant_signal:" + connectionId
func participantSignalKey(connectionID livekit.ConnectionID) string {
return "participant_signal:" + string(connectionID)
}
func rtcNodeChannel(nodeId string) string {
return "rtc_channel:" + nodeId
func rtcNodeChannel(nodeID livekit.NodeID) string {
return "rtc_channel:" + string(nodeID)
}
func signalNodeChannel(nodeId string) string {
return "signal_channel:" + nodeId
func signalNodeChannel(nodeID livekit.NodeID) string {
return "signal_channel:" + string(nodeID)
}
func publishRTCMessage(rc *redis.Client, nodeId string, participantKey string, msg proto.Message) error {
func publishRTCMessage(rc *redis.Client, nodeID livekit.NodeID, participantKey livekit.ParticipantKey, msg proto.Message) error {
rm := &livekit.RTCNodeMessage{
ParticipantKey: participantKey,
ParticipantKey: string(participantKey),
}
switch o := msg.(type) {
case *livekit.StartSession:
@@ -52,7 +52,7 @@ func publishRTCMessage(rc *redis.Client, nodeId string, participantKey string, m
}
case *livekit.RTCNodeMessage:
rm = o
rm.ParticipantKey = participantKey
rm.ParticipantKey = string(participantKey)
default:
return ErrInvalidRouterMessage
}
@@ -61,14 +61,14 @@ func publishRTCMessage(rc *redis.Client, nodeId string, participantKey string, m
return err
}
// logger.Debugw("publishing to rtc", "rtcChannel", rtcNodeChannel(nodeId),
// logger.Debugw("publishing to rtc", "rtcChannel", rtcNodeChannel(nodeID),
// "message", rm.Message)
return rc.Publish(redisCtx, rtcNodeChannel(nodeId), data).Err()
return rc.Publish(redisCtx, rtcNodeChannel(nodeID), data).Err()
}
func publishSignalMessage(rc *redis.Client, nodeId string, connectionId string, msg proto.Message) error {
func publishSignalMessage(rc *redis.Client, nodeID livekit.NodeID, connectionID livekit.ConnectionID, msg proto.Message) error {
rm := &livekit.SignalNodeMessage{
ConnectionId: connectionId,
ConnectionId: string(connectionID),
}
switch o := msg.(type) {
case *livekit.SignalResponse:
@@ -87,23 +87,23 @@ func publishSignalMessage(rc *redis.Client, nodeId string, connectionId string,
return err
}
// logger.Debugw("publishing to signal", "signalChannel", signalNodeChannel(nodeId),
// logger.Debugw("publishing to signal", "signalChannel", signalNodeChannel(nodeID),
// "message", rm.Message)
return rc.Publish(redisCtx, signalNodeChannel(nodeId), data).Err()
return rc.Publish(redisCtx, signalNodeChannel(nodeID), data).Err()
}
type RTCNodeSink struct {
rc *redis.Client
nodeId string
participantKey string
nodeID livekit.NodeID
participantKey livekit.ParticipantKey
isClosed utils.AtomicFlag
onClose func()
}
func NewRTCNodeSink(rc *redis.Client, nodeId, participantKey string) *RTCNodeSink {
func NewRTCNodeSink(rc *redis.Client, nodeID livekit.NodeID, participantKey livekit.ParticipantKey) *RTCNodeSink {
return &RTCNodeSink{
rc: rc,
nodeId: nodeId,
nodeID: nodeID,
participantKey: participantKey,
}
}
@@ -112,7 +112,7 @@ func (s *RTCNodeSink) WriteMessage(msg proto.Message) error {
if s.isClosed.Get() {
return ErrChannelClosed
}
return publishRTCMessage(s.rc, s.nodeId, s.participantKey, msg)
return publishRTCMessage(s.rc, s.nodeID, s.participantKey, msg)
}
func (s *RTCNodeSink) Close() {
@@ -130,17 +130,17 @@ func (s *RTCNodeSink) OnClose(f func()) {
type SignalNodeSink struct {
rc *redis.Client
nodeId string
connectionId string
nodeID livekit.NodeID
connectionID livekit.ConnectionID
isClosed utils.AtomicFlag
onClose func()
}
func NewSignalNodeSink(rc *redis.Client, nodeId, connectionId string) *SignalNodeSink {
func NewSignalNodeSink(rc *redis.Client, nodeID livekit.NodeID, connectionID livekit.ConnectionID) *SignalNodeSink {
return &SignalNodeSink{
rc: rc,
nodeId: nodeId,
connectionId: connectionId,
nodeID: nodeID,
connectionID: connectionID,
}
}
@@ -148,14 +148,14 @@ func (s *SignalNodeSink) WriteMessage(msg proto.Message) error {
if s.isClosed.Get() {
return ErrChannelClosed
}
return publishSignalMessage(s.rc, s.nodeId, s.connectionId, msg)
return publishSignalMessage(s.rc, s.nodeID, s.connectionID, msg)
}
func (s *SignalNodeSink) Close() {
if !s.isClosed.TrySet(true) {
return
}
_ = publishSignalMessage(s.rc, s.nodeId, s.connectionId, &livekit.EndSession{})
_ = publishSignalMessage(s.rc, s.nodeID, s.connectionID, &livekit.EndSession{})
if s.onClose != nil {
s.onClose()
}
pkg/routing/redisrouter.go
+35 -35
View File
@@ -85,11 +85,11 @@ func (r *RedisRouter) GetNodeForRoom(_ context.Context, roomName livekit.RoomNam
return nil, errors.Wrap(err, "could not get node for room")
}
return r.GetNode(nodeID)
return r.GetNode(livekit.NodeID(nodeID))
}
func (r *RedisRouter) SetNodeForRoom(_ context.Context, roomName livekit.RoomName, nodeID string) error {
return r.rc.HSet(r.ctx, NodeRoomKey, string(roomName), nodeID).Err()
func (r *RedisRouter) SetNodeForRoom(_ context.Context, roomName livekit.RoomName, nodeID livekit.NodeID) error {
return r.rc.HSet(r.ctx, NodeRoomKey, string(roomName), string(nodeID)).Err()
}
func (r *RedisRouter) ClearRoomState(_ context.Context, roomName livekit.RoomName) error {
@@ -99,8 +99,8 @@ func (r *RedisRouter) ClearRoomState(_ context.Context, roomName livekit.RoomNam
return nil
}
func (r *RedisRouter) GetNode(nodeID string) (*livekit.Node, error) {
data, err := r.rc.HGet(r.ctx, NodesKey, nodeID).Result()
func (r *RedisRouter) GetNode(nodeID livekit.NodeID) (*livekit.Node, error) {
data, err := r.rc.HGet(r.ctx, NodesKey, string(nodeID)).Result()
if err == redis.Nil {
return nil, ErrNotFound
} else if err != nil {
@@ -130,7 +130,7 @@ func (r *RedisRouter) ListNodes() ([]*livekit.Node, error) {
}
// StartParticipantSignal signal connection sets up paths to the RTC node, and starts to route messages to that message queue
func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (connectionID string, reqSink MessageSink, resSource MessageSource, err error) {
func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (connectionID livekit.ConnectionID, reqSink MessageSink, resSource MessageSource, err error) {
// find the node where the room is hosted at
rtcNode, err := r.GetNodeForRoom(ctx, roomName)
if err != nil {
@@ -138,7 +138,7 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livek
}
// create a new connection id
connectionID = utils.NewGuid("CO_")
connectionID = livekit.ConnectionID(utils.NewGuid("CO_"))
pKey := participantKey(roomName, pi.Identity)
// map signal & rtc nodes
@@ -146,7 +146,7 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livek
return
}
sink := NewRTCNodeSink(r.rc, rtcNode.Id, pKey)
sink := NewRTCNodeSink(r.rc, livekit.NodeID(rtcNode.Id), pKey)
// serialize claims
claims, err := json.Marshal(pi.Grants)
@@ -161,7 +161,7 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livek
Metadata: pi.Metadata,
Name: string(pi.Name),
// connection id is to allow the RTC node to identify where to route the message back to
ConnectionId: connectionID,
ConnectionId: string(connectionID),
Reconnect: pi.Reconnect,
Permission: pi.Permission,
AutoSubscribe: pi.AutoSubscribe,
@@ -174,8 +174,8 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livek
return
}
// index by connectionId, since there may be multiple connections for the participant
resChan := r.getOrCreateMessageChannel(r.responseChannels, connectionID)
// index by connectionID, since there may be multiple connections for the participant
resChan := r.getOrCreateMessageChannel(r.responseChannels, string(connectionID))
return connectionID, sink, resChan, nil
}
@@ -186,8 +186,8 @@ func (r *RedisRouter) WriteParticipantRTC(_ context.Context, roomName livekit.Ro
return err
}
rtcSink := NewRTCNodeSink(r.rc, rtcNode, pkey)
msg.ParticipantKey = participantKey(roomName, identity)
rtcSink := NewRTCNodeSink(r.rc, livekit.NodeID(rtcNode), pkey)
msg.ParticipantKey = string(participantKey(roomName, identity))
return r.writeRTCMessage(rtcSink, msg)
}
@@ -196,16 +196,16 @@ func (r *RedisRouter) WriteRoomRTC(ctx context.Context, roomName livekit.RoomNam
if err != nil {
return err
}
msg.ParticipantKey = participantKey(roomName, "")
msg.ParticipantKey = string(participantKey(roomName, ""))
return r.WriteNodeRTC(ctx, node.Id, msg)
}
func (r *RedisRouter) WriteNodeRTC(_ context.Context, rtcNodeID string, msg *livekit.RTCNodeMessage) error {
rtcSink := NewRTCNodeSink(r.rc, rtcNodeID, msg.ParticipantKey)
rtcSink := NewRTCNodeSink(r.rc, livekit.NodeID(rtcNodeID), livekit.ParticipantKey(msg.ParticipantKey))
return r.writeRTCMessage(rtcSink, msg)
}
func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantKey string) error {
func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantKey livekit.ParticipantKey) error {
// find the node where the room is hosted at
rtcNode, err := r.GetNodeForRoom(r.ctx, livekit.RoomName(ss.RoomName))
if err != nil {
@@ -225,7 +225,7 @@ func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantK
}
// find signal node to send responses back
signalNode, err := r.getParticipantSignalNode(ss.ConnectionId)
signalNode, err := r.getParticipantSignalNode(livekit.ConnectionID(ss.ConnectionId))
if err != nil {
return err
}
@@ -240,7 +240,7 @@ func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantK
// the previous rtc worker thread is still consuming off of it.
// we'll want to sever the connection and switch to the new one
r.lock.RLock()
requestChan, ok := r.requestChannels[participantKey]
requestChan, ok := r.requestChannels[string(participantKey)]
r.lock.RUnlock()
if ok {
requestChan.Close()
@@ -265,8 +265,8 @@ func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantK
Grants: claims,
}
reqChan := r.getOrCreateMessageChannel(r.requestChannels, participantKey)
resSink := NewSignalNodeSink(r.rc, signalNode, ss.ConnectionId)
reqChan := r.getOrCreateMessageChannel(r.requestChannels, string(participantKey))
resSink := NewSignalNodeSink(r.rc, livekit.NodeID(signalNode), livekit.ConnectionID(ss.ConnectionId))
r.onNewParticipant(
r.ctx,
livekit.RoomName(ss.RoomName),
@@ -312,22 +312,22 @@ func (r *RedisRouter) Stop() {
r.cancel()
}
func (r *RedisRouter) setParticipantRTCNode(participantKey, nodeId string) error {
err := r.rc.Set(r.ctx, participantRTCKey(participantKey), nodeId, participantMappingTTL).Err()
func (r *RedisRouter) setParticipantRTCNode(participantKey livekit.ParticipantKey, nodeID string) error {
err := r.rc.Set(r.ctx, participantRTCKey(participantKey), nodeID, participantMappingTTL).Err()
if err != nil {
err = errors.Wrap(err, "could not set rtc node")
}
return err
}
func (r *RedisRouter) setParticipantSignalNode(connectionId, nodeId string) error {
if err := r.rc.Set(r.ctx, participantSignalKey(connectionId), nodeId, participantMappingTTL).Err(); err != nil {
func (r *RedisRouter) setParticipantSignalNode(connectionID livekit.ConnectionID, nodeID string) error {
if err := r.rc.Set(r.ctx, participantSignalKey(connectionID), nodeID, participantMappingTTL).Err(); err != nil {
return errors.Wrap(err, "could not set signal node")
}
return nil
}
func (r *RedisRouter) getParticipantRTCNode(participantKey string) (string, error) {
func (r *RedisRouter) getParticipantRTCNode(participantKey livekit.ParticipantKey) (string, error) {
val, err := r.rc.Get(r.ctx, participantRTCKey(participantKey)).Result()
if err == redis.Nil {
err = ErrNodeNotFound
@@ -335,8 +335,8 @@ func (r *RedisRouter) getParticipantRTCNode(participantKey string) (string, erro
return val, err
}
func (r *RedisRouter) getParticipantSignalNode(connectionId string) (nodeId string, err error) {
val, err := r.rc.Get(r.ctx, participantSignalKey(connectionId)).Result()
func (r *RedisRouter) getParticipantSignalNode(connectionID livekit.ConnectionID) (nodeID string, err error) {
val, err := r.rc.Get(r.ctx, participantSignalKey(connectionID)).Result()
if err == redis.Nil {
err = ErrNodeNotFound
}
@@ -366,8 +366,8 @@ func (r *RedisRouter) redisWorker(startedChan chan struct{}) {
}()
logger.Debugw("starting redisWorker", "nodeID", r.currentNode.Id)
sigChannel := signalNodeChannel(r.currentNode.Id)
rtcChannel := rtcNodeChannel(r.currentNode.Id)
sigChannel := signalNodeChannel(livekit.NodeID(r.currentNode.Id))
rtcChannel := rtcNodeChannel(livekit.NodeID(r.currentNode.Id))
r.pubsub = r.rc.Subscribe(r.ctx, sigChannel, rtcChannel)
close(startedChan)
@@ -407,10 +407,10 @@ func (r *RedisRouter) redisWorker(startedChan chan struct{}) {
}
func (r *RedisRouter) handleSignalMessage(sm *livekit.SignalNodeMessage) error {
connectionId := sm.ConnectionId
connectionID := sm.ConnectionId
r.lock.RLock()
resSink := r.responseChannels[connectionId]
resSink := r.responseChannels[connectionID]
r.lock.RUnlock()
// if a client closed the channel, then sent more messages after that,
@@ -421,7 +421,7 @@ func (r *RedisRouter) handleSignalMessage(sm *livekit.SignalNodeMessage) error {
switch rmb := sm.Message.(type) {
case *livekit.SignalNodeMessage_Response:
// logger.Debugw("forwarding signal message",
// "connID", connectionId,
// "connID", connectionID,
// "type", fmt.Sprintf("%T", rmb.Response.Message))
if err := resSink.WriteMessage(rmb.Response); err != nil {
return err
@@ -429,14 +429,14 @@ func (r *RedisRouter) handleSignalMessage(sm *livekit.SignalNodeMessage) error {
case *livekit.SignalNodeMessage_EndSession:
// logger.Debugw("received EndSession, closing signal connection",
// "connID", connectionId)
// "connID", connectionID)
resSink.Close()
}
return nil
}
func (r *RedisRouter) handleRTCMessage(rm *livekit.RTCNodeMessage) error {
pKey := rm.ParticipantKey
pKey := livekit.ParticipantKey(rm.ParticipantKey)
switch rmb := rm.Message.(type) {
case *livekit.RTCNodeMessage_StartSession:
@@ -447,7 +447,7 @@ func (r *RedisRouter) handleRTCMessage(rm *livekit.RTCNodeMessage) error {
case *livekit.RTCNodeMessage_Request:
r.lock.RLock()
requestChan := r.requestChannels[pKey]
requestChan := r.requestChannels[string(pKey)]
r.lock.RUnlock()
if requestChan == nil {
return ErrChannelClosed
pkg/routing/routingfakes/fake_router.go
+16 -16
View File
@@ -82,12 +82,12 @@ type FakeRouter struct {
removeDeadNodesReturnsOnCall map[int]struct {
result1 error
}
SetNodeForRoomStub func(context.Context, livekit.RoomName, string) error
SetNodeForRoomStub func(context.Context, livekit.RoomName, livekit.NodeID) error
setNodeForRoomMutex sync.RWMutex
setNodeForRoomArgsForCall []struct {
arg1 context.Context
arg2 livekit.RoomName
arg3 string
arg3 livekit.NodeID
}
setNodeForRoomReturns struct {
result1 error
@@ -105,7 +105,7 @@ type FakeRouter struct {
startReturnsOnCall map[int]struct {
result1 error
}
StartParticipantSignalStub func(context.Context, livekit.RoomName, routing.ParticipantInit) (string, routing.MessageSink, routing.MessageSource, error)
StartParticipantSignalStub func(context.Context, livekit.RoomName, routing.ParticipantInit) (livekit.ConnectionID, routing.MessageSink, routing.MessageSource, error)
startParticipantSignalMutex sync.RWMutex
startParticipantSignalArgsForCall []struct {
arg1 context.Context
@@ -113,13 +113,13 @@ type FakeRouter struct {
arg3 routing.ParticipantInit
}
startParticipantSignalReturns struct {
result1 string
result1 livekit.ConnectionID
result2 routing.MessageSink
result3 routing.MessageSource
result4 error
}
startParticipantSignalReturnsOnCall map[int]struct {
result1 string
result1 livekit.ConnectionID
result2 routing.MessageSink
result3 routing.MessageSource
result4 error
@@ -546,13 +546,13 @@ func (fake *FakeRouter) RemoveDeadNodesReturnsOnCall(i int, result1 error) {
}{result1}
}
func (fake *FakeRouter) SetNodeForRoom(arg1 context.Context, arg2 livekit.RoomName, arg3 string) error {
func (fake *FakeRouter) SetNodeForRoom(arg1 context.Context, arg2 livekit.RoomName, arg3 livekit.NodeID) error {
fake.setNodeForRoomMutex.Lock()
ret, specificReturn := fake.setNodeForRoomReturnsOnCall[len(fake.setNodeForRoomArgsForCall)]
fake.setNodeForRoomArgsForCall = append(fake.setNodeForRoomArgsForCall, struct {
arg1 context.Context
arg2 livekit.RoomName
arg3 string
arg3 livekit.NodeID
}{arg1, arg2, arg3})
stub := fake.SetNodeForRoomStub
fakeReturns := fake.setNodeForRoomReturns
@@ -573,13 +573,13 @@ func (fake *FakeRouter) SetNodeForRoomCallCount() int {
return len(fake.setNodeForRoomArgsForCall)
}
func (fake *FakeRouter) SetNodeForRoomCalls(stub func(context.Context, livekit.RoomName, string) error) {
func (fake *FakeRouter) SetNodeForRoomCalls(stub func(context.Context, livekit.RoomName, livekit.NodeID) error) {
fake.setNodeForRoomMutex.Lock()
defer fake.setNodeForRoomMutex.Unlock()
fake.SetNodeForRoomStub = stub
}
func (fake *FakeRouter) SetNodeForRoomArgsForCall(i int) (context.Context, livekit.RoomName, string) {
func (fake *FakeRouter) SetNodeForRoomArgsForCall(i int) (context.Context, livekit.RoomName, livekit.NodeID) {
fake.setNodeForRoomMutex.RLock()
defer fake.setNodeForRoomMutex.RUnlock()
argsForCall := fake.setNodeForRoomArgsForCall[i]
@@ -662,7 +662,7 @@ func (fake *FakeRouter) StartReturnsOnCall(i int, result1 error) {
}{result1}
}
func (fake *FakeRouter) StartParticipantSignal(arg1 context.Context, arg2 livekit.RoomName, arg3 routing.ParticipantInit) (string, routing.MessageSink, routing.MessageSource, error) {
func (fake *FakeRouter) StartParticipantSignal(arg1 context.Context, arg2 livekit.RoomName, arg3 routing.ParticipantInit) (livekit.ConnectionID, routing.MessageSink, routing.MessageSource, error) {
fake.startParticipantSignalMutex.Lock()
ret, specificReturn := fake.startParticipantSignalReturnsOnCall[len(fake.startParticipantSignalArgsForCall)]
fake.startParticipantSignalArgsForCall = append(fake.startParticipantSignalArgsForCall, struct {
@@ -689,7 +689,7 @@ func (fake *FakeRouter) StartParticipantSignalCallCount() int {
return len(fake.startParticipantSignalArgsForCall)
}
func (fake *FakeRouter) StartParticipantSignalCalls(stub func(context.Context, livekit.RoomName, routing.ParticipantInit) (string, routing.MessageSink, routing.MessageSource, error)) {
func (fake *FakeRouter) StartParticipantSignalCalls(stub func(context.Context, livekit.RoomName, routing.ParticipantInit) (livekit.ConnectionID, routing.MessageSink, routing.MessageSource, error)) {
fake.startParticipantSignalMutex.Lock()
defer fake.startParticipantSignalMutex.Unlock()
fake.StartParticipantSignalStub = stub
@@ -702,32 +702,32 @@ func (fake *FakeRouter) StartParticipantSignalArgsForCall(i int) (context.Contex
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
}
func (fake *FakeRouter) StartParticipantSignalReturns(result1 string, result2 routing.MessageSink, result3 routing.MessageSource, result4 error) {
func (fake *FakeRouter) StartParticipantSignalReturns(result1 livekit.ConnectionID, result2 routing.MessageSink, result3 routing.MessageSource, result4 error) {
fake.startParticipantSignalMutex.Lock()
defer fake.startParticipantSignalMutex.Unlock()
fake.StartParticipantSignalStub = nil
fake.startParticipantSignalReturns = struct {
result1 string
result1 livekit.ConnectionID
result2 routing.MessageSink
result3 routing.MessageSource
result4 error
}{result1, result2, result3, result4}
}
func (fake *FakeRouter) StartParticipantSignalReturnsOnCall(i int, result1 string, result2 routing.MessageSink, result3 routing.MessageSource, result4 error) {
func (fake *FakeRouter) StartParticipantSignalReturnsOnCall(i int, result1 livekit.ConnectionID, result2 routing.MessageSink, result3 routing.MessageSource, result4 error) {
fake.startParticipantSignalMutex.Lock()
defer fake.startParticipantSignalMutex.Unlock()
fake.StartParticipantSignalStub = nil
if fake.startParticipantSignalReturnsOnCall == nil {
fake.startParticipantSignalReturnsOnCall = make(map[int]struct {
result1 string
result1 livekit.ConnectionID
result2 routing.MessageSink
result3 routing.MessageSource
result4 error
})
}
fake.startParticipantSignalReturnsOnCall[i] = struct {
result1 string
result1 livekit.ConnectionID
result2 routing.MessageSink
result3 routing.MessageSource
result4 error
pkg/routing/utils.go
+4 -4
View File
@@ -7,12 +7,12 @@ import (
"github.com/livekit/protocol/livekit"
)
func participantKey(roomName livekit.RoomName, identity livekit.ParticipantIdentity) string {
return string(roomName) + "|" + string(identity)
func participantKey(roomName livekit.RoomName, identity livekit.ParticipantIdentity) livekit.ParticipantKey {
return livekit.ParticipantKey(string(roomName) + "|" + string(identity))
}
func parseParticipantKey(pkey string) (roomName livekit.RoomName, identity livekit.ParticipantIdentity, err error) {
parts := strings.Split(pkey, "|")
func parseParticipantKey(pkey livekit.ParticipantKey) (roomName livekit.RoomName, identity livekit.ParticipantIdentity, err error) {
parts := strings.Split(string(pkey), "|")
if len(parts) != 2 {
err = errors.New("invalid participant key")
return
pkg/rtc/mediatrackreceiver.go
+1 -1
View File
@@ -304,7 +304,7 @@ func (t *MediaTrackReceiver) handleMaxLossFeedback(_ *sfu.DownTrack, report *rtc
t.maybeUpdateLoss()
}
func (t *MediaTrackReceiver) NotifySubscriberNodeMediaLoss(_nodeID string, fractionalLoss uint8) {
func (t *MediaTrackReceiver) NotifySubscriberNodeMediaLoss(_nodeID livekit.NodeID, fractionalLoss uint8) {
t.downFracLostLock.Lock()
if t.maxDownFracLost < fractionalLoss {
t.maxDownFracLost = fractionalLoss
pkg/rtc/mediatracksubscriptions.go
+3 -3
View File
@@ -34,7 +34,7 @@ type MediaTrackSubscriptions struct {
// quality level enable/disable
maxQualityLock sync.RWMutex
maxSubscriberQuality map[livekit.ParticipantID]livekit.VideoQuality
maxSubscriberNodeQuality map[string]livekit.VideoQuality // nodeID => livekit.VideoQuality
maxSubscriberNodeQuality map[livekit.NodeID]livekit.VideoQuality
maxSubscribedQuality livekit.VideoQuality
onSubscribedMaxQualityChange func(subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality)
maxQualityTimer *time.Timer
@@ -57,7 +57,7 @@ func NewMediaTrackSubscriptions(params MediaTrackSubscriptionsParams) *MediaTrac
params: params,
subscribedTracks: make(map[livekit.ParticipantID]types.SubscribedTrack),
maxSubscriberQuality: make(map[livekit.ParticipantID]livekit.VideoQuality),
maxSubscriberNodeQuality: make(map[string]livekit.VideoQuality),
maxSubscriberNodeQuality: make(map[livekit.NodeID]livekit.VideoQuality),
}
return t
@@ -433,7 +433,7 @@ func (t *MediaTrackSubscriptions) notifySubscriberMaxQuality(subscriberID liveki
t.UpdateQualityChange(false)
}
func (t *MediaTrackSubscriptions) NotifySubscriberNodeMaxQuality(nodeID string, quality livekit.VideoQuality) {
func (t *MediaTrackSubscriptions) NotifySubscriberNodeMaxQuality(nodeID livekit.NodeID, quality livekit.VideoQuality) {
if t.params.MediaTrack.Kind() != livekit.TrackType_VIDEO {
return
}
pkg/rtc/types/interfaces.go
+4 -4
View File
@@ -75,8 +75,8 @@ type Participant interface {
// updates from remotes
UpdateSubscriptionPermission(subscriptionPermission *livekit.SubscriptionPermission, resolver func(participantID livekit.ParticipantID) LocalParticipant) error
UpdateVideoLayers(updateVideoLayers *livekit.UpdateVideoLayers) error
UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error
UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error
UpdateSubscribedQuality(nodeID livekit.NodeID, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error
UpdateMediaLoss(nodeID livekit.NodeID, trackID livekit.TrackID, fractionalLoss uint32) error
DebugInfo() map[string]interface{}
}
@@ -204,8 +204,8 @@ type MediaTrack interface {
// returns quality information that's appropriate for width & height
GetQualityForDimension(width, height uint32) livekit.VideoQuality
NotifySubscriberNodeMaxQuality(nodeID string, quality livekit.VideoQuality)
NotifySubscriberNodeMediaLoss(nodeID string, fractionalLoss uint8)
NotifySubscriberNodeMaxQuality(nodeID livekit.NodeID, quality livekit.VideoQuality)
NotifySubscriberNodeMediaLoss(nodeID livekit.NodeID, fractionalLoss uint8)
}
//counterfeiter:generate . LocalMediaTrack
pkg/rtc/types/typesfakes/fake_local_media_track.go
+12 -12
View File
@@ -121,16 +121,16 @@ type FakeLocalMediaTrack struct {
nameReturnsOnCall map[int]struct {
result1 string
}
NotifySubscriberNodeMaxQualityStub func(string, livekit.VideoQuality)
NotifySubscriberNodeMaxQualityStub func(livekit.NodeID, livekit.VideoQuality)
notifySubscriberNodeMaxQualityMutex sync.RWMutex
notifySubscriberNodeMaxQualityArgsForCall []struct {
arg1 string
arg1 livekit.NodeID
arg2 livekit.VideoQuality
}
NotifySubscriberNodeMediaLossStub func(string, uint8)
NotifySubscriberNodeMediaLossStub func(livekit.NodeID, uint8)
notifySubscriberNodeMediaLossMutex sync.RWMutex
notifySubscriberNodeMediaLossArgsForCall []struct {
arg1 string
arg1 livekit.NodeID
arg2 uint8
}
PublisherIDStub func() livekit.ParticipantID
@@ -833,10 +833,10 @@ func (fake *FakeLocalMediaTrack) NameReturnsOnCall(i int, result1 string) {
}{result1}
}
func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQuality(arg1 string, arg2 livekit.VideoQuality) {
func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQuality(arg1 livekit.NodeID, arg2 livekit.VideoQuality) {
fake.notifySubscriberNodeMaxQualityMutex.Lock()
fake.notifySubscriberNodeMaxQualityArgsForCall = append(fake.notifySubscriberNodeMaxQualityArgsForCall, struct {
arg1 string
arg1 livekit.NodeID
arg2 livekit.VideoQuality
}{arg1, arg2})
stub := fake.NotifySubscriberNodeMaxQualityStub
@@ -853,23 +853,23 @@ func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQualityCallCount() int {
return len(fake.notifySubscriberNodeMaxQualityArgsForCall)
}
func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQualityCalls(stub func(string, livekit.VideoQuality)) {
func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQualityCalls(stub func(livekit.NodeID, livekit.VideoQuality)) {
fake.notifySubscriberNodeMaxQualityMutex.Lock()
defer fake.notifySubscriberNodeMaxQualityMutex.Unlock()
fake.NotifySubscriberNodeMaxQualityStub = stub
}
func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQualityArgsForCall(i int) (string, livekit.VideoQuality) {
func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMaxQualityArgsForCall(i int) (livekit.NodeID, livekit.VideoQuality) {
fake.notifySubscriberNodeMaxQualityMutex.RLock()
defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock()
argsForCall := fake.notifySubscriberNodeMaxQualityArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2
}
func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMediaLoss(arg1 string, arg2 uint8) {
func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMediaLoss(arg1 livekit.NodeID, arg2 uint8) {
fake.notifySubscriberNodeMediaLossMutex.Lock()
fake.notifySubscriberNodeMediaLossArgsForCall = append(fake.notifySubscriberNodeMediaLossArgsForCall, struct {
arg1 string
arg1 livekit.NodeID
arg2 uint8
}{arg1, arg2})
stub := fake.NotifySubscriberNodeMediaLossStub
@@ -886,13 +886,13 @@ func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMediaLossCallCount() int {
return len(fake.notifySubscriberNodeMediaLossArgsForCall)
}
func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMediaLossCalls(stub func(string, uint8)) {
func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMediaLossCalls(stub func(livekit.NodeID, uint8)) {
fake.notifySubscriberNodeMediaLossMutex.Lock()
defer fake.notifySubscriberNodeMediaLossMutex.Unlock()
fake.NotifySubscriberNodeMediaLossStub = stub
}
func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMediaLossArgsForCall(i int) (string, uint8) {
func (fake *FakeLocalMediaTrack) NotifySubscriberNodeMediaLossArgsForCall(i int) (livekit.NodeID, uint8) {
fake.notifySubscriberNodeMediaLossMutex.RLock()
defer fake.notifySubscriberNodeMediaLossMutex.RUnlock()
argsForCall := fake.notifySubscriberNodeMediaLossArgsForCall[i]
pkg/rtc/types/typesfakes/fake_local_participant.go
+12 -12
View File
@@ -537,10 +537,10 @@ type FakeLocalParticipant struct {
toProtoReturnsOnCall map[int]struct {
result1 *livekit.ParticipantInfo
}
UpdateMediaLossStub func(string, livekit.TrackID, uint32) error
UpdateMediaLossStub func(livekit.NodeID, livekit.TrackID, uint32) error
updateMediaLossMutex sync.RWMutex
updateMediaLossArgsForCall []struct {
arg1 string
arg1 livekit.NodeID
arg2 livekit.TrackID
arg3 uint32
}
@@ -555,10 +555,10 @@ type FakeLocalParticipant struct {
updateRTTArgsForCall []struct {
arg1 uint32
}
UpdateSubscribedQualityStub func(string, livekit.TrackID, livekit.VideoQuality) error
UpdateSubscribedQualityStub func(livekit.NodeID, livekit.TrackID, livekit.VideoQuality) error
updateSubscribedQualityMutex sync.RWMutex
updateSubscribedQualityArgsForCall []struct {
arg1 string
arg1 livekit.NodeID
arg2 livekit.TrackID
arg3 livekit.VideoQuality
}
@@ -3474,11 +3474,11 @@ func (fake *FakeLocalParticipant) ToProtoReturnsOnCall(i int, result1 *livekit.P
}{result1}
}
func (fake *FakeLocalParticipant) UpdateMediaLoss(arg1 string, arg2 livekit.TrackID, arg3 uint32) error {
func (fake *FakeLocalParticipant) UpdateMediaLoss(arg1 livekit.NodeID, arg2 livekit.TrackID, arg3 uint32) error {
fake.updateMediaLossMutex.Lock()
ret, specificReturn := fake.updateMediaLossReturnsOnCall[len(fake.updateMediaLossArgsForCall)]
fake.updateMediaLossArgsForCall = append(fake.updateMediaLossArgsForCall, struct {
arg1 string
arg1 livekit.NodeID
arg2 livekit.TrackID
arg3 uint32
}{arg1, arg2, arg3})
@@ -3501,13 +3501,13 @@ func (fake *FakeLocalParticipant) UpdateMediaLossCallCount() int {
return len(fake.updateMediaLossArgsForCall)
}
func (fake *FakeLocalParticipant) UpdateMediaLossCalls(stub func(string, livekit.TrackID, uint32) error) {
func (fake *FakeLocalParticipant) UpdateMediaLossCalls(stub func(livekit.NodeID, livekit.TrackID, uint32) error) {
fake.updateMediaLossMutex.Lock()
defer fake.updateMediaLossMutex.Unlock()
fake.UpdateMediaLossStub = stub
}
func (fake *FakeLocalParticipant) UpdateMediaLossArgsForCall(i int) (string, livekit.TrackID, uint32) {
func (fake *FakeLocalParticipant) UpdateMediaLossArgsForCall(i int) (livekit.NodeID, livekit.TrackID, uint32) {
fake.updateMediaLossMutex.RLock()
defer fake.updateMediaLossMutex.RUnlock()
argsForCall := fake.updateMediaLossArgsForCall[i]
@@ -3569,11 +3569,11 @@ func (fake *FakeLocalParticipant) UpdateRTTArgsForCall(i int) uint32 {
return argsForCall.arg1
}
func (fake *FakeLocalParticipant) UpdateSubscribedQuality(arg1 string, arg2 livekit.TrackID, arg3 livekit.VideoQuality) error {
func (fake *FakeLocalParticipant) UpdateSubscribedQuality(arg1 livekit.NodeID, arg2 livekit.TrackID, arg3 livekit.VideoQuality) error {
fake.updateSubscribedQualityMutex.Lock()
ret, specificReturn := fake.updateSubscribedQualityReturnsOnCall[len(fake.updateSubscribedQualityArgsForCall)]
fake.updateSubscribedQualityArgsForCall = append(fake.updateSubscribedQualityArgsForCall, struct {
arg1 string
arg1 livekit.NodeID
arg2 livekit.TrackID
arg3 livekit.VideoQuality
}{arg1, arg2, arg3})
@@ -3596,13 +3596,13 @@ func (fake *FakeLocalParticipant) UpdateSubscribedQualityCallCount() int {
return len(fake.updateSubscribedQualityArgsForCall)
}
func (fake *FakeLocalParticipant) UpdateSubscribedQualityCalls(stub func(string, livekit.TrackID, livekit.VideoQuality) error) {
func (fake *FakeLocalParticipant) UpdateSubscribedQualityCalls(stub func(livekit.NodeID, livekit.TrackID, livekit.VideoQuality) error) {
fake.updateSubscribedQualityMutex.Lock()
defer fake.updateSubscribedQualityMutex.Unlock()
fake.UpdateSubscribedQualityStub = stub
}
func (fake *FakeLocalParticipant) UpdateSubscribedQualityArgsForCall(i int) (string, livekit.TrackID, livekit.VideoQuality) {
func (fake *FakeLocalParticipant) UpdateSubscribedQualityArgsForCall(i int) (livekit.NodeID, livekit.TrackID, livekit.VideoQuality) {
fake.updateSubscribedQualityMutex.RLock()
defer fake.updateSubscribedQualityMutex.RUnlock()
argsForCall := fake.updateSubscribedQualityArgsForCall[i]
pkg/rtc/types/typesfakes/fake_media_track.go
+12 -12
View File
@@ -99,16 +99,16 @@ type FakeMediaTrack struct {
nameReturnsOnCall map[int]struct {
result1 string
}
NotifySubscriberNodeMaxQualityStub func(string, livekit.VideoQuality)
NotifySubscriberNodeMaxQualityStub func(livekit.NodeID, livekit.VideoQuality)
notifySubscriberNodeMaxQualityMutex sync.RWMutex
notifySubscriberNodeMaxQualityArgsForCall []struct {
arg1 string
arg1 livekit.NodeID
arg2 livekit.VideoQuality
}
NotifySubscriberNodeMediaLossStub func(string, uint8)
NotifySubscriberNodeMediaLossStub func(livekit.NodeID, uint8)
notifySubscriberNodeMediaLossMutex sync.RWMutex
notifySubscriberNodeMediaLossArgsForCall []struct {
arg1 string
arg1 livekit.NodeID
arg2 uint8
}
PublisherIDStub func() livekit.ParticipantID
@@ -677,10 +677,10 @@ func (fake *FakeMediaTrack) NameReturnsOnCall(i int, result1 string) {
}{result1}
}
func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQuality(arg1 string, arg2 livekit.VideoQuality) {
func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQuality(arg1 livekit.NodeID, arg2 livekit.VideoQuality) {
fake.notifySubscriberNodeMaxQualityMutex.Lock()
fake.notifySubscriberNodeMaxQualityArgsForCall = append(fake.notifySubscriberNodeMaxQualityArgsForCall, struct {
arg1 string
arg1 livekit.NodeID
arg2 livekit.VideoQuality
}{arg1, arg2})
stub := fake.NotifySubscriberNodeMaxQualityStub
@@ -697,23 +697,23 @@ func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityCallCount() int {
return len(fake.notifySubscriberNodeMaxQualityArgsForCall)
}
func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityCalls(stub func(string, livekit.VideoQuality)) {
func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityCalls(stub func(livekit.NodeID, livekit.VideoQuality)) {
fake.notifySubscriberNodeMaxQualityMutex.Lock()
defer fake.notifySubscriberNodeMaxQualityMutex.Unlock()
fake.NotifySubscriberNodeMaxQualityStub = stub
}
func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityArgsForCall(i int) (string, livekit.VideoQuality) {
func (fake *FakeMediaTrack) NotifySubscriberNodeMaxQualityArgsForCall(i int) (livekit.NodeID, livekit.VideoQuality) {
fake.notifySubscriberNodeMaxQualityMutex.RLock()
defer fake.notifySubscriberNodeMaxQualityMutex.RUnlock()
argsForCall := fake.notifySubscriberNodeMaxQualityArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2
}
func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLoss(arg1 string, arg2 uint8) {
func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLoss(arg1 livekit.NodeID, arg2 uint8) {
fake.notifySubscriberNodeMediaLossMutex.Lock()
fake.notifySubscriberNodeMediaLossArgsForCall = append(fake.notifySubscriberNodeMediaLossArgsForCall, struct {
arg1 string
arg1 livekit.NodeID
arg2 uint8
}{arg1, arg2})
stub := fake.NotifySubscriberNodeMediaLossStub
@@ -730,13 +730,13 @@ func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossCallCount() int {
return len(fake.notifySubscriberNodeMediaLossArgsForCall)
}
func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossCalls(stub func(string, uint8)) {
func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossCalls(stub func(livekit.NodeID, uint8)) {
fake.notifySubscriberNodeMediaLossMutex.Lock()
defer fake.notifySubscriberNodeMediaLossMutex.Unlock()
fake.NotifySubscriberNodeMediaLossStub = stub
}
func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossArgsForCall(i int) (string, uint8) {
func (fake *FakeMediaTrack) NotifySubscriberNodeMediaLossArgsForCall(i int) (livekit.NodeID, uint8) {
fake.notifySubscriberNodeMediaLossMutex.RLock()
defer fake.notifySubscriberNodeMediaLossMutex.RUnlock()
argsForCall := fake.notifySubscriberNodeMediaLossArgsForCall[i]
pkg/rtc/types/typesfakes/fake_participant.go
+12 -12
View File
@@ -141,10 +141,10 @@ type FakeParticipant struct {
toProtoReturnsOnCall map[int]struct {
result1 *livekit.ParticipantInfo
}
UpdateMediaLossStub func(string, livekit.TrackID, uint32) error
UpdateMediaLossStub func(livekit.NodeID, livekit.TrackID, uint32) error
updateMediaLossMutex sync.RWMutex
updateMediaLossArgsForCall []struct {
arg1 string
arg1 livekit.NodeID
arg2 livekit.TrackID
arg3 uint32
}
@@ -154,10 +154,10 @@ type FakeParticipant struct {
updateMediaLossReturnsOnCall map[int]struct {
result1 error
}
UpdateSubscribedQualityStub func(string, livekit.TrackID, livekit.VideoQuality) error
UpdateSubscribedQualityStub func(livekit.NodeID, livekit.TrackID, livekit.VideoQuality) error
updateSubscribedQualityMutex sync.RWMutex
updateSubscribedQualityArgsForCall []struct {
arg1 string
arg1 livekit.NodeID
arg2 livekit.TrackID
arg3 livekit.VideoQuality
}
@@ -895,11 +895,11 @@ func (fake *FakeParticipant) ToProtoReturnsOnCall(i int, result1 *livekit.Partic
}{result1}
}
func (fake *FakeParticipant) UpdateMediaLoss(arg1 string, arg2 livekit.TrackID, arg3 uint32) error {
func (fake *FakeParticipant) UpdateMediaLoss(arg1 livekit.NodeID, arg2 livekit.TrackID, arg3 uint32) error {
fake.updateMediaLossMutex.Lock()
ret, specificReturn := fake.updateMediaLossReturnsOnCall[len(fake.updateMediaLossArgsForCall)]
fake.updateMediaLossArgsForCall = append(fake.updateMediaLossArgsForCall, struct {
arg1 string
arg1 livekit.NodeID
arg2 livekit.TrackID
arg3 uint32
}{arg1, arg2, arg3})
@@ -922,13 +922,13 @@ func (fake *FakeParticipant) UpdateMediaLossCallCount() int {
return len(fake.updateMediaLossArgsForCall)
}
func (fake *FakeParticipant) UpdateMediaLossCalls(stub func(string, livekit.TrackID, uint32) error) {
func (fake *FakeParticipant) UpdateMediaLossCalls(stub func(livekit.NodeID, livekit.TrackID, uint32) error) {
fake.updateMediaLossMutex.Lock()
defer fake.updateMediaLossMutex.Unlock()
fake.UpdateMediaLossStub = stub
}
func (fake *FakeParticipant) UpdateMediaLossArgsForCall(i int) (string, livekit.TrackID, uint32) {
func (fake *FakeParticipant) UpdateMediaLossArgsForCall(i int) (livekit.NodeID, livekit.TrackID, uint32) {
fake.updateMediaLossMutex.RLock()
defer fake.updateMediaLossMutex.RUnlock()
argsForCall := fake.updateMediaLossArgsForCall[i]
@@ -958,11 +958,11 @@ func (fake *FakeParticipant) UpdateMediaLossReturnsOnCall(i int, result1 error)
}{result1}
}
func (fake *FakeParticipant) UpdateSubscribedQuality(arg1 string, arg2 livekit.TrackID, arg3 livekit.VideoQuality) error {
func (fake *FakeParticipant) UpdateSubscribedQuality(arg1 livekit.NodeID, arg2 livekit.TrackID, arg3 livekit.VideoQuality) error {
fake.updateSubscribedQualityMutex.Lock()
ret, specificReturn := fake.updateSubscribedQualityReturnsOnCall[len(fake.updateSubscribedQualityArgsForCall)]
fake.updateSubscribedQualityArgsForCall = append(fake.updateSubscribedQualityArgsForCall, struct {
arg1 string
arg1 livekit.NodeID
arg2 livekit.TrackID
arg3 livekit.VideoQuality
}{arg1, arg2, arg3})
@@ -985,13 +985,13 @@ func (fake *FakeParticipant) UpdateSubscribedQualityCallCount() int {
return len(fake.updateSubscribedQualityArgsForCall)
}
func (fake *FakeParticipant) UpdateSubscribedQualityCalls(stub func(string, livekit.TrackID, livekit.VideoQuality) error) {
func (fake *FakeParticipant) UpdateSubscribedQualityCalls(stub func(livekit.NodeID, livekit.TrackID, livekit.VideoQuality) error) {
fake.updateSubscribedQualityMutex.Lock()
defer fake.updateSubscribedQualityMutex.Unlock()
fake.UpdateSubscribedQualityStub = stub
}
func (fake *FakeParticipant) UpdateSubscribedQualityArgsForCall(i int) (string, livekit.TrackID, livekit.VideoQuality) {
func (fake *FakeParticipant) UpdateSubscribedQualityArgsForCall(i int) (livekit.NodeID, livekit.TrackID, livekit.VideoQuality) {
fake.updateSubscribedQualityMutex.RLock()
defer fake.updateSubscribedQualityMutex.RUnlock()
argsForCall := fake.updateSubscribedQualityArgsForCall[i]
pkg/rtc/uptrackmanager.go
+2 -2
View File
@@ -222,7 +222,7 @@ func (u *UpTrackManager) UpdateVideoLayers(updateVideoLayers *livekit.UpdateVide
return nil
}
func (u *UpTrackManager) UpdateSubscribedQuality(nodeID string, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error {
func (u *UpTrackManager) UpdateSubscribedQuality(nodeID livekit.NodeID, trackID livekit.TrackID, maxQuality livekit.VideoQuality) error {
track := u.GetPublishedTrack(trackID)
if track == nil {
u.params.Logger.Warnw("could not find track", nil, "trackID", trackID)
@@ -233,7 +233,7 @@ func (u *UpTrackManager) UpdateSubscribedQuality(nodeID string, trackID livekit.
return nil
}
func (u *UpTrackManager) UpdateMediaLoss(nodeID string, trackID livekit.TrackID, fractionalLoss uint32) error {
func (u *UpTrackManager) UpdateMediaLoss(nodeID livekit.NodeID, trackID livekit.TrackID, fractionalLoss uint32) error {
track := u.GetPublishedTrack(trackID)
if track == nil {
u.params.Logger.Warnw("could not find track", nil, "trackID", trackID)
pkg/service/roomallocator.go
+2 -2
View File
@@ -89,7 +89,7 @@ func (r *StandardRoomAllocator) CreateRoom(ctx context.Context, req *livekit.Cre
}
// select a new node
nodeID := req.NodeId
nodeID := livekit.NodeID(req.NodeId)
if nodeID == "" {
nodes, err := r.router.ListNodes()
if err != nil {
@@ -101,7 +101,7 @@ func (r *StandardRoomAllocator) CreateRoom(ctx context.Context, req *livekit.Cre
return nil, err
}
nodeID = node.Id
nodeID = livekit.NodeID(node.Id)
}
logger.Debugw("selected node for room", "room", rm.Name, "roomID", rm.Sid, "nodeID", nodeID)