Help us shape the future of Go! Take the 2026 Go Developer Survey

routing

package
v1.13.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 18, 2026 License: Apache-2.0 Imports: 25 Imported by: 12

Documentation

Index

Constants

View Source 
const (
	// hash of node_id => Node proto
	NodesKey = "nodes"

	// hash of room_name => node_id
	NodeRoomKey = "room_node_map"
)
View Source 
const DefaultMessageChannelSize = 200

Variables

View Source 
var (
	ErrNotFound             = errors.New("could not find object")
	ErrIPNotSet             = errors.New("ip address is required and not set")
	ErrHandlerNotDefined    = errors.New("handler not defined")
	ErrIncorrectRTCNode     = errors.New("current node isn't the RTC node for the room")
	ErrNodeNotFound         = errors.New("could not locate the node")
	ErrNodeLimitReached     = errors.New("reached configured limit for node")
	ErrInvalidRouterMessage = errors.New("invalid router message")
	ErrChannelClosed        = errors.New("channel closed")
	ErrChannelFull          = errors.New("channel is full")

	// errors when starting signal connection
	ErrRequestChannelClosed       = errors.New("request channel closed")
	ErrCouldNotMigrateParticipant = errors.New("could not migrate participant")
	ErrClientInfoNotSet           = errors.New("client info not set")
)
View Source 
var ErrSignalMessageDropped = errors.New("signal message dropped")
View Source 
var ErrSignalWriteFailed = errors.New("signal write failed")

Functions

func CopySignalStreamToMessageChannel added in v1.4.2 

func CopySignalStreamToMessageChannel[SendType, RecvType RelaySignalMessage](
	stream psrpc.Stream[SendType, RecvType],
	ch *MessageChannel,
	reader SignalMessageReader[RecvType],
	config config.SignalRelayConfig,
	promSignalSuccess func(),
	promSignalFailure func(),
) error

Types

type LocalNode 

type LocalNode interface {
	Clone() *livekit.Node
	SetNodeID(nodeID livekit.NodeID)
	NodeID() livekit.NodeID
	NodeType() livekit.NodeType
	NodeIP() string
	Region() string
	SetState(state livekit.NodeState)
	SetStats(stats *livekit.NodeStats)
	UpdateNodeStats() bool
	SecondsSinceNodeStatsUpdate() float64
}

type LocalNodeImpl added in v1.8.1 

type LocalNodeImpl struct {
	// contains filtered or unexported fields
}

func NewLocalNode 

func NewLocalNode(conf *config.Config) (*LocalNodeImpl, error)

func NewLocalNodeFromNodeProto added in v1.8.1 

func NewLocalNodeFromNodeProto(node *livekit.Node) (*LocalNodeImpl, error)

func (*LocalNodeImpl) Clone added in v1.8.1 

func (l *LocalNodeImpl) Clone() *livekit.Node

func (*LocalNodeImpl) NodeID added in v1.8.1 

func (l *LocalNodeImpl) NodeID() livekit.NodeID

func (*LocalNodeImpl) NodeIP added in v1.8.1 

func (l *LocalNodeImpl) NodeIP() string

func (*LocalNodeImpl) NodeType added in v1.8.1 

func (l *LocalNodeImpl) NodeType() livekit.NodeType

func (*LocalNodeImpl) Region added in v1.8.1 

func (l *LocalNodeImpl) Region() string

func (*LocalNodeImpl) SecondsSinceNodeStatsUpdate added in v1.8.1 

func (l *LocalNodeImpl) SecondsSinceNodeStatsUpdate() float64

func (*LocalNodeImpl) SetNodeID added in v1.8.1 

func (l *LocalNodeImpl) SetNodeID(nodeID livekit.NodeID)

for testing only

func (*LocalNodeImpl) SetState added in v1.8.1 

func (l *LocalNodeImpl) SetState(state livekit.NodeState)

func (*LocalNodeImpl) SetStats added in v1.8.1 

func (l *LocalNodeImpl) SetStats(stats *livekit.NodeStats)

for testing only

func (*LocalNodeImpl) UpdateNodeStats added in v1.8.1 

func (l *LocalNodeImpl) UpdateNodeStats() bool

type LocalRouter 

type LocalRouter struct {
	// contains filtered or unexported fields
}

a router of messages on the same node, basic implementation for local testing

func NewLocalRouter 

func NewLocalRouter(
	currentNode LocalNode,
	signalClient SignalClient,
	roomManagerClient RoomManagerClient,
	nodeStatsConfig config.NodeStatsConfig,
) *LocalRouter

func (*LocalRouter) ClearRoomState 

func (r *LocalRouter) ClearRoomState(_ context.Context, _ livekit.RoomName) error

func (*LocalRouter) CreateRoom added in v1.8.0 

func (r *LocalRouter) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (res *livekit.Room, err error)

func (*LocalRouter) CreateRoomWithNodeID added in v1.8.0 

func (r *LocalRouter) CreateRoomWithNodeID(ctx context.Context, req *livekit.CreateRoomRequest, nodeID livekit.NodeID) (res *livekit.Room, err error)

func (*LocalRouter) Drain added in v0.13.5 

func (r *LocalRouter) Drain()

func (*LocalRouter) GetNode 

func (r *LocalRouter) GetNode(nodeID livekit.NodeID) (*livekit.Node, error)

func (*LocalRouter) GetNodeForRoom 

func (r *LocalRouter) GetNodeForRoom(_ context.Context, _ livekit.RoomName) (*livekit.Node, error)

func (*LocalRouter) GetRegion added in v0.15.7 

func (r *LocalRouter) GetRegion() string

func (*LocalRouter) ListNodes 

func (r *LocalRouter) ListNodes() ([]*livekit.Node, error)

func (*LocalRouter) RegisterNode 

func (r *LocalRouter) RegisterNode() error

func (*LocalRouter) RemoveDeadNodes 

func (r *LocalRouter) RemoveDeadNodes() error

func (*LocalRouter) SetNodeForRoom 

func (r *LocalRouter) SetNodeForRoom(_ context.Context, _ livekit.RoomName, _ livekit.NodeID) error

func (*LocalRouter) Start 

func (r *LocalRouter) Start() error

func (*LocalRouter) StartParticipantSignal 

func (r *LocalRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (res StartParticipantSignalResults, err error)

func (*LocalRouter) StartParticipantSignalWithNodeID added in v1.4.0 

func (r *LocalRouter) StartParticipantSignalWithNodeID(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit, nodeID livekit.NodeID) (res StartParticipantSignalResults, err error)

func (*LocalRouter) Stop 

func (r *LocalRouter) Stop()

func (*LocalRouter) UnregisterNode 

func (r *LocalRouter) UnregisterNode() error

type MessageChannel 

type MessageChannel struct {
	// contains filtered or unexported fields
}

func NewDefaultMessageChannel added in v1.4.0 

func NewDefaultMessageChannel(connectionID livekit.ConnectionID) *MessageChannel

func NewMessageChannel 

func NewMessageChannel(connectionID livekit.ConnectionID, size int) *MessageChannel

func (*MessageChannel) Close 

func (m *MessageChannel) Close()

func (*MessageChannel) ConnectionID added in v1.4.4 

func (m *MessageChannel) ConnectionID() livekit.ConnectionID

func (*MessageChannel) IsClosed added in v0.13.1 

func (m *MessageChannel) IsClosed() bool

func (*MessageChannel) OnClose 

func (m *MessageChannel) OnClose(f func())

func (*MessageChannel) ReadChan 

func (m *MessageChannel) ReadChan() <-chan proto.Message

func (*MessageChannel) WriteMessage 

func (m *MessageChannel) WriteMessage(msg proto.Message) error

type MessageRouter added in v0.14.2 

type MessageRouter interface {
	// CreateRoom starts an rtc room
	CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (res *livekit.Room, err error)

	// StartParticipantSignal participant signal connection is ready to start
	StartParticipantSignal(
		ctx context.Context,
		roomName livekit.RoomName,
		pi ParticipantInit,
	) (res StartParticipantSignalResults, err error)
}

type MessageSink 

type MessageSink interface {
	WriteMessage(msg proto.Message) error
	IsClosed() bool
	Close()
	ConnectionID() livekit.ConnectionID
}

MessageSink is an abstraction for writing protobuf messages and having them read by a MessageSource, potentially on a different node via a transport

func NewSignalMessageSink added in v1.4.2 

func NewSignalMessageSink[SendType, RecvType RelaySignalMessage](params SignalSinkParams[SendType, RecvType]) MessageSink

type MessageSource 

type MessageSource interface {
	// ReadChan exposes a one way channel to make it easier to use with select
	ReadChan() <-chan proto.Message
	IsClosed() bool
	Close()
	ConnectionID() livekit.ConnectionID
}

type NodeStats 

type NodeStats struct {
	// contains filtered or unexported fields
}

func NewNodeStats added in v1.9.0 

func NewNodeStats(conf *config.NodeStatsConfig, startedAt int64) *NodeStats

func (*NodeStats) GetLatestNodeStats added in v1.13.2 

func (n *NodeStats) GetLatestNodeStats() *livekit.NodeStats

func (*NodeStats) UpdateAndGetNodeStats added in v1.9.0 

func (n *NodeStats) UpdateAndGetNodeStats() (*livekit.NodeStats, error)

func (*NodeStats) UpdateConfig added in v1.9.0 

func (n *NodeStats) UpdateConfig(conf *config.NodeStatsConfig)

type NullMessageSink added in v1.8.1 

type NullMessageSink struct {
	// contains filtered or unexported fields
}

func NewNullMessageSink added in v1.8.1 

func NewNullMessageSink(connID livekit.ConnectionID) *NullMessageSink

func (*NullMessageSink) Close added in v1.8.1 

func (n *NullMessageSink) Close()

func (*NullMessageSink) ConnectionID added in v1.8.1 

func (n *NullMessageSink) ConnectionID() livekit.ConnectionID

func (*NullMessageSink) IsClosed added in v1.8.1 

func (n *NullMessageSink) IsClosed() bool

func (*NullMessageSink) WriteMessage added in v1.8.1 

func (n *NullMessageSink) WriteMessage(_msg proto.Message) error

type NullMessageSource added in v1.8.1 

type NullMessageSource struct {
	// contains filtered or unexported fields
}

func NewNullMessageSource added in v1.8.1 

func NewNullMessageSource(connID livekit.ConnectionID) *NullMessageSource

func (*NullMessageSource) Close added in v1.8.1 

func (n *NullMessageSource) Close()

func (*NullMessageSource) ConnectionID added in v1.8.1 

func (n *NullMessageSource) ConnectionID() livekit.ConnectionID

func (*NullMessageSource) IsClosed added in v1.8.1 

func (n *NullMessageSource) IsClosed() bool

func (*NullMessageSource) ReadChan added in v1.8.1 

func (n *NullMessageSource) ReadChan() <-chan proto.Message

type ParticipantInit 

type ParticipantInit struct {
	Identity                livekit.ParticipantIdentity
	Name                    livekit.ParticipantName
	Reconnect               bool
	ReconnectReason         livekit.ReconnectReason
	AutoSubscribe           bool
	AutoSubscribeDataTrack  *bool
	Client                  *livekit.ClientInfo
	Grants                  *auth.ClaimGrants
	TokenExpiresAt          time.Time
	Region                  string
	AdaptiveStream          bool
	ID                      livekit.ParticipantID
	SubscriberAllowPause    *bool
	DisableICELite          bool
	CreateRoom              *livekit.CreateRoomRequest
	AddTrackRequests        []*livekit.AddTrackRequest
	PublisherOffer          *livekit.SessionDescription
	SyncState               *livekit.SyncState
	UseSinglePeerConnection bool
}

func ParticipantInitFromStartSession added in v0.15.7 

func ParticipantInitFromStartSession(ss *livekit.StartSession, region string) (*ParticipantInit, error)

func (*ParticipantInit) MarshalLogObject added in v1.8.1 

func (pi *ParticipantInit) MarshalLogObject(e zapcore.ObjectEncoder) error

func (*ParticipantInit) ToStartSession added in v0.15.7 

func (pi *ParticipantInit) ToStartSession(roomName livekit.RoomName, connectionID livekit.ConnectionID) (*livekit.StartSession, error)

type RedisRouter 

type RedisRouter struct {
	*LocalRouter
	// contains filtered or unexported fields
}

RedisRouter uses Redis pub/sub to route signaling messages across different nodes It relies on the RTC node to be the primary driver of the participant connection. Because

func NewRedisRouter 

func NewRedisRouter(lr *LocalRouter, rc redis.UniversalClient, kps rpc.KeepalivePubSub) *RedisRouter

func (*RedisRouter) ClearRoomState 

func (r *RedisRouter) ClearRoomState(_ context.Context, roomName livekit.RoomName) error

func (*RedisRouter) CreateRoom added in v1.8.0 

func (r *RedisRouter) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (res *livekit.Room, err error)

func (*RedisRouter) Drain added in v0.13.5 

func (r *RedisRouter) Drain()

func (*RedisRouter) GetNode 

func (r *RedisRouter) GetNode(nodeID livekit.NodeID) (*livekit.Node, error)

func (*RedisRouter) GetNodeForRoom 

func (r *RedisRouter) GetNodeForRoom(_ context.Context, roomName livekit.RoomName) (*livekit.Node, error)

GetNodeForRoom finds the node where the room is hosted at

func (*RedisRouter) ListNodes 

func (r *RedisRouter) ListNodes() ([]*livekit.Node, error)

func (*RedisRouter) RegisterNode 

func (r *RedisRouter) RegisterNode() error

func (*RedisRouter) RemoveDeadNodes 

func (r *RedisRouter) RemoveDeadNodes() error

func (*RedisRouter) SetNodeForRoom 

func (r *RedisRouter) SetNodeForRoom(_ context.Context, roomName livekit.RoomName, nodeID livekit.NodeID) error

func (*RedisRouter) Start 

func (r *RedisRouter) Start() error

func (*RedisRouter) StartParticipantSignal 

func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (res StartParticipantSignalResults, err error)

StartParticipantSignal signal connection sets up paths to the RTC node, and starts to route messages to that message queue

func (*RedisRouter) Stop 

func (r *RedisRouter) Stop()

func (*RedisRouter) UnregisterNode 

func (r *RedisRouter) UnregisterNode() error

type RelaySignalMessage added in v1.4.2 

type RelaySignalMessage interface {
	proto.Message
	GetSeq() uint64
	GetClose() bool
}

type RoomManagerClient added in v1.8.0 

type RoomManagerClient interface {
	rpc.TypedRoomManagerClient
}

func NewRoomManagerClient added in v1.8.0 

func NewRoomManagerClient(clientParams rpc.ClientParams, config config.RoomConfig) (RoomManagerClient, error)

type Router 

type Router interface {
	MessageRouter

	RegisterNode() error
	UnregisterNode() error
	RemoveDeadNodes() error

	ListNodes() ([]*livekit.Node, error)

	GetNodeForRoom(ctx context.Context, roomName livekit.RoomName) (*livekit.Node, error)
	SetNodeForRoom(ctx context.Context, roomName livekit.RoomName, nodeId livekit.NodeID) error
	ClearRoomState(ctx context.Context, roomName livekit.RoomName) error

	GetRegion() string

	Start() error
	Drain()
	Stop()
}

Router allows multiple nodes to coordinate the participant session

func CreateRouter added in v0.13.5 

func CreateRouter(
	rc redis.UniversalClient,
	node LocalNode,
	signalClient SignalClient,
	roomManagerClient RoomManagerClient,
	kps rpc.KeepalivePubSub,
	nodeStatsConfig config.NodeStatsConfig,
) Router

type SignalClient added in v1.4.0 

type SignalClient interface {
	ActiveCount() int
	StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit, nodeID livekit.NodeID) (connectionID livekit.ConnectionID, reqSink MessageSink, resSource MessageSource, err error)
}

func NewSignalClient added in v1.4.0 

func NewSignalClient(nodeID livekit.NodeID, bus psrpc.MessageBus, config config.SignalRelayConfig) (SignalClient, error)

type SignalMessageReader added in v1.4.2 

type SignalMessageReader[RecvType RelaySignalMessage] interface {
	Read(msg RecvType) ([]proto.Message, error)
}

type SignalMessageWriter added in v1.4.2 

type SignalMessageWriter[SendType RelaySignalMessage] interface {
	Write(seq uint64, close bool, msgs []proto.Message) SendType
}

type SignalSinkParams added in v1.4.2 

type SignalSinkParams[SendType, RecvType RelaySignalMessage] struct {
	Stream         psrpc.Stream[SendType, RecvType]
	Logger         logger.Logger
	Config         config.SignalRelayConfig
	Writer         SignalMessageWriter[SendType]
	CloseOnFailure bool
	BlockOnClose   bool
	ConnectionID   livekit.ConnectionID
}

type StartParticipantSignalResults added in v1.5.2 

type StartParticipantSignalResults struct {
	ConnectionID        livekit.ConnectionID
	RequestSink         MessageSink
	ResponseSource      MessageSource
	NodeID              livekit.NodeID
	NodeSelectionReason string
}

Source Files

View all Source files

Directories

Path Synopsis
Code generated by counterfeiter.
 Code generated by counterfeiter.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.