From 3f381721393ef9996f58011cc2c23d690c641900 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Mon, 21 Feb 2022 08:52:11 +0530 Subject: [PATCH] Types for ConnectionID, NodeID, ParticipantKey (#450) * Types for ConnectionID, NodeID, ParticipantKey * latest protocol version --- go.mod | 2 +- go.sum | 4 +- pkg/routing/interfaces.go | 4 +- pkg/routing/localrouter.go | 24 +++---- pkg/routing/redis.go | 58 +++++++-------- pkg/routing/redisrouter.go | 70 +++++++++---------- pkg/routing/routingfakes/fake_router.go | 32 ++++----- pkg/routing/utils.go | 8 +-- pkg/rtc/mediatrackreceiver.go | 2 +- pkg/rtc/mediatracksubscriptions.go | 6 +- pkg/rtc/types/interfaces.go | 8 +-- .../typesfakes/fake_local_media_track.go | 24 +++---- .../typesfakes/fake_local_participant.go | 24 +++---- pkg/rtc/types/typesfakes/fake_media_track.go | 24 +++---- pkg/rtc/types/typesfakes/fake_participant.go | 24 +++---- pkg/rtc/uptrackmanager.go | 4 +- pkg/service/roomallocator.go | 4 +- 17 files changed, 161 insertions(+), 161 deletions(-) diff --git a/go.mod b/go.mod index a0a23b579..1c841aaa3 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.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 diff --git a/go.sum b/go.sum index 573facb69..4a0d1c723 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.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= diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index 7708be14c..20c4bc4f5 100644 --- a/pkg/routing/interfaces.go +++ b/pkg/routing/interfaces.go @@ -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 diff --git a/pkg/routing/localrouter.go b/pkg/routing/localrouter.go index d054dc5a1..73aace1b6 100644 --- a/pkg/routing/localrouter.go +++ b/pkg/routing/localrouter.go @@ -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 diff --git a/pkg/routing/redis.go b/pkg/routing/redis.go index a1c36ab62..4098476fb 100644 --- a/pkg/routing/redis.go +++ b/pkg/routing/redis.go @@ -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() } diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index d6632e476..d34cf5f1f 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -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 diff --git a/pkg/routing/routingfakes/fake_router.go b/pkg/routing/routingfakes/fake_router.go index 423441dec..0826793cb 100644 --- a/pkg/routing/routingfakes/fake_router.go +++ b/pkg/routing/routingfakes/fake_router.go @@ -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 diff --git a/pkg/routing/utils.go b/pkg/routing/utils.go index 0da5cda8a..7e0ac9388 100644 --- a/pkg/routing/utils.go +++ b/pkg/routing/utils.go @@ -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 diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index 887b895b3..f3e7cd1eb 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -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 diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index c7b509574..9f74c18e9 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -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 } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 46fbd7524..2d75fe9d4 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -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 diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index f98ff878b..86d083a2c 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -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] diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index cb59677d2..7fb0fe01b 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -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] diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index 0cc53898c..5172162fd 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -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] diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 7f7718008..dbba2c0a6 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -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] diff --git a/pkg/rtc/uptrackmanager.go b/pkg/rtc/uptrackmanager.go index f81815d82..377e3f0fc 100644 --- a/pkg/rtc/uptrackmanager.go +++ b/pkg/rtc/uptrackmanager.go @@ -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) diff --git a/pkg/service/roomallocator.go b/pkg/service/roomallocator.go index 9dd3cf2e0..a960e7840 100644 --- a/pkg/service/roomallocator.go +++ b/pkg/service/roomallocator.go @@ -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)