service

package
v1.4.3-dev5 Latest Latest
Warning

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

Go to latest
Published: May 19, 2023 License: Apache-2.0 Imports: 57 Imported by: 0

Documentation

Index

Constants

View Source
const (
	VersionKey = "livekit_version"

	// RoomsKey is hash of room_name => Room proto
	RoomsKey        = "rooms"
	RoomInternalKey = "room_internal"

	// EgressKey is a hash of egressID => egress info
	EgressKey        = "egress"
	EndedEgressKey   = "ended_egress"
	RoomEgressPrefix = "egress:room:"

	// IngressKey is a hash of ingressID => ingress info
	IngressKey         = "ingress"
	StreamKeyKey       = "{ingress}_stream_key"
	IngressStatePrefix = "{ingress}_state:"
	RoomIngressPrefix  = "room_{ingress}:"

	// RoomParticipantsPrefix is hash of participant_name => ParticipantInfo
	RoomParticipantsPrefix = "room_participants:"

	// RoomLockPrefix is a simple key containing a provided lock uid
	RoomLockPrefix = "room_lock:"
)
View Source
const (
	LivekitRealm = "livekit"
)

Variables

View Source
var (
	ErrPermissionDenied          = errors.New("permissions denied")
	ErrMissingAuthorization      = errors.New("invalid authorization header. Must start with " + bearerPrefix)
	ErrInvalidAuthorizationToken = errors.New("invalid authorization token")
)
View Source
var (
	ErrEgressNotFound        = psrpc.NewErrorf(psrpc.NotFound, "egress does not exist")
	ErrEgressNotConnected    = psrpc.NewErrorf(psrpc.Internal, "egress not connected (redis required)")
	ErrIdentityEmpty         = psrpc.NewErrorf(psrpc.InvalidArgument, "identity cannot be empty")
	ErrIngressNotConnected   = psrpc.NewErrorf(psrpc.Internal, "ingress not connected (redis required)")
	ErrIngressNotFound       = psrpc.NewErrorf(psrpc.NotFound, "ingress does not exist")
	ErrMetadataExceedsLimits = psrpc.NewErrorf(psrpc.InvalidArgument, "metadata size exceeds limits")
	ErrOperationFailed       = psrpc.NewErrorf(psrpc.Internal, "operation cannot be completed")
	ErrParticipantNotFound   = psrpc.NewErrorf(psrpc.NotFound, "participant does not exist")
	ErrRoomNotFound          = psrpc.NewErrorf(psrpc.NotFound, "requested room does not exist")
	ErrRoomLockFailed        = psrpc.NewErrorf(psrpc.Internal, "could not lock room")
	ErrRoomUnlockFailed      = psrpc.NewErrorf(psrpc.Internal, "could not unlock room, lock token does not match")
	ErrTrackNotFound         = psrpc.NewErrorf(psrpc.NotFound, "track is not found")
	ErrWebHookMissingAPIKey  = psrpc.NewErrorf(psrpc.InvalidArgument, "api_key is required to use webhooks")
)

Functions

func AppendLogFields

func AppendLogFields(ctx context.Context, fields ...interface{})

func EnsureAdminPermission

func EnsureAdminPermission(ctx context.Context, room livekit.RoomName) error

func EnsureCreatePermission

func EnsureCreatePermission(ctx context.Context) error

func EnsureIngressAdminPermission

func EnsureIngressAdminPermission(ctx context.Context) error

func EnsureJoinPermission

func EnsureJoinPermission(ctx context.Context) (name livekit.RoomName, err error)

func EnsureListPermission

func EnsureListPermission(ctx context.Context) error

func EnsureRecordPermission

func EnsureRecordPermission(ctx context.Context) error

func GetClientIP

func GetClientIP(r *http.Request) string

func GetGrants

func GetGrants(ctx context.Context) *auth.ClaimGrants

func InitializeRouter

func InitializeRouter(conf *config.Config, currentNode routing.LocalNode) (routing.Router, error)

func IsValidDomain

func IsValidDomain(domain string) bool

func NewTurnServer

func NewTurnServer(conf *config.Config, authHandler turn.AuthHandler, standalone bool) (*turn.Server, error)

func SetAuthorizationToken

func SetAuthorizationToken(r *http.Request, token string)

func TwirpLogger

func TwirpLogger(logger logger.Logger) *twirp.ServerHooks

logging handling inspired by https://github.com/bakins/twirpzap License: Apache-2.0

func TwirpRequestStatusReporter

func TwirpRequestStatusReporter() *twirp.ServerHooks

func WithGrants

func WithGrants(ctx context.Context, grants *auth.ClaimGrants) context.Context

Types

type APIKeyAuthMiddleware

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

authentication middleware

func NewAPIKeyAuthMiddleware

func NewAPIKeyAuthMiddleware(provider auth.KeyProvider) *APIKeyAuthMiddleware

func (*APIKeyAuthMiddleware) ServeHTTP

type EgressService

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

func (*EgressService) ListEgress

func (*EgressService) StartRoomCompositeEgress

func (s *EgressService) StartRoomCompositeEgress(ctx context.Context, req *livekit.RoomCompositeEgressRequest) (*livekit.EgressInfo, error)

func (*EgressService) StartTrackCompositeEgress

func (s *EgressService) StartTrackCompositeEgress(ctx context.Context, req *livekit.TrackCompositeEgressRequest) (*livekit.EgressInfo, error)

func (*EgressService) StartTrackEgress

func (s *EgressService) StartTrackEgress(ctx context.Context, req *livekit.TrackEgressRequest) (*livekit.EgressInfo, error)

func (*EgressService) StartWebEgress

func (s *EgressService) StartWebEgress(ctx context.Context, req *livekit.WebEgressRequest) (*livekit.EgressInfo, error)

func (*EgressService) StopEgress

func (*EgressService) UpdateLayout

func (*EgressService) UpdateStream

type EgressStore

type EgressStore interface {
	StoreEgress(ctx context.Context, info *livekit.EgressInfo) error
	LoadEgress(ctx context.Context, egressID string) (*livekit.EgressInfo, error)
	ListEgress(ctx context.Context, roomName livekit.RoomName, active bool) ([]*livekit.EgressInfo, error)
	UpdateEgress(ctx context.Context, info *livekit.EgressInfo) error
}

type IOInfoService

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

func (*IOInfoService) GetIngressInfo

func (*IOInfoService) Start

func (s *IOInfoService) Start() error

func (*IOInfoService) Stop

func (s *IOInfoService) Stop()

func (*IOInfoService) UpdateEgressInfo

func (s *IOInfoService) UpdateEgressInfo(ctx context.Context, info *livekit.EgressInfo) (*emptypb.Empty, error)

func (*IOInfoService) UpdateIngressState

func (s *IOInfoService) UpdateIngressState(ctx context.Context, req *rpc.UpdateIngressStateRequest) (*emptypb.Empty, error)

type IngressService

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

func (*IngressService) CreateIngress

func (*IngressService) CreateIngressWithUrlPrefix

func (s *IngressService) CreateIngressWithUrlPrefix(ctx context.Context, urlPrefix string, req *livekit.CreateIngressRequest) (*livekit.IngressInfo, error)

func (*IngressService) DeleteIngress

func (*IngressService) ListIngress

func (*IngressService) UpdateIngress

type IngressStore

type IngressStore interface {
	StoreIngress(ctx context.Context, info *livekit.IngressInfo) error
	LoadIngress(ctx context.Context, ingressID string) (*livekit.IngressInfo, error)
	LoadIngressFromStreamKey(ctx context.Context, streamKey string) (*livekit.IngressInfo, error)
	ListIngress(ctx context.Context, roomName livekit.RoomName) ([]*livekit.IngressInfo, error)
	UpdateIngress(ctx context.Context, info *livekit.IngressInfo) error
	UpdateIngressState(ctx context.Context, ingressId string, state *livekit.IngressState) error
	DeleteIngress(ctx context.Context, info *livekit.IngressInfo) error
}

type LayoutMetadata

type LayoutMetadata struct {
	Layout string `json:"layout"`
}

type LivekitServer

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

func InitializeServer

func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*LivekitServer, error)

func NewLivekitServer

func NewLivekitServer(conf *config.Config,
	roomService livekit.RoomService,
	egressService *EgressService,
	ingressService *IngressService,
	ioService *IOInfoService,
	rtcService *RTCService,
	keyProvider auth.KeyProvider,
	router routing.Router,
	roomManager *RoomManager,
	signalServer *SignalServer,
	turnServer *turn.Server,
	currentNode routing.LocalNode,
) (s *LivekitServer, err error)

func (*LivekitServer) HTTPPort

func (s *LivekitServer) HTTPPort() int

func (*LivekitServer) IsRunning

func (s *LivekitServer) IsRunning() bool

func (*LivekitServer) Node

func (s *LivekitServer) Node() *livekit.Node

func (*LivekitServer) RoomManager

func (s *LivekitServer) RoomManager() *RoomManager

func (*LivekitServer) Start

func (s *LivekitServer) Start() error

func (*LivekitServer) Stop

func (s *LivekitServer) Stop(force bool)

type LocalStore

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

encapsulates CRUD operations for room settings

func NewLocalStore

func NewLocalStore() *LocalStore

func (*LocalStore) DeleteParticipant

func (s *LocalStore) DeleteParticipant(_ context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity) error

func (*LocalStore) DeleteRoom

func (s *LocalStore) DeleteRoom(ctx context.Context, roomName livekit.RoomName) error

func (*LocalStore) ListParticipants

func (s *LocalStore) ListParticipants(_ context.Context, roomName livekit.RoomName) ([]*livekit.ParticipantInfo, error)

func (*LocalStore) ListRooms

func (s *LocalStore) ListRooms(_ context.Context, roomNames []livekit.RoomName) ([]*livekit.Room, error)

func (*LocalStore) LoadParticipant

func (s *LocalStore) LoadParticipant(_ context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity) (*livekit.ParticipantInfo, error)

func (*LocalStore) LoadRoom

func (s *LocalStore) LoadRoom(_ context.Context, roomName livekit.RoomName, includeInternal bool) (*livekit.Room, *livekit.RoomInternal, error)

func (*LocalStore) LockRoom

func (*LocalStore) StoreParticipant

func (s *LocalStore) StoreParticipant(_ context.Context, roomName livekit.RoomName, participant *livekit.ParticipantInfo) error

func (*LocalStore) StoreRoom

func (s *LocalStore) StoreRoom(_ context.Context, room *livekit.Room, internal *livekit.RoomInternal) error

func (*LocalStore) UnlockRoom

func (s *LocalStore) UnlockRoom(_ context.Context, _ livekit.RoomName, _ string) error

type ObjectStore

type ObjectStore interface {
	ServiceStore

	// enable locking on a specific room to prevent race
	// returns a (lock uuid, error)
	LockRoom(ctx context.Context, roomName livekit.RoomName, duration time.Duration) (string, error)
	UnlockRoom(ctx context.Context, roomName livekit.RoomName, uid string) error

	StoreRoom(ctx context.Context, room *livekit.Room, internal *livekit.RoomInternal) error
	DeleteRoom(ctx context.Context, roomName livekit.RoomName) error

	StoreParticipant(ctx context.Context, roomName livekit.RoomName, participant *livekit.ParticipantInfo) error
	DeleteParticipant(ctx context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity) error
}

encapsulates CRUD operations for room settings

type RTCService

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

func NewRTCService

func NewRTCService(
	conf *config.Config,
	ra RoomAllocator,
	store ServiceStore,
	router routing.MessageRouter,
	currentNode routing.LocalNode,
	telemetry telemetry.TelemetryService,
) *RTCService

func (*RTCService) ParseClientInfo

func (s *RTCService) ParseClientInfo(r *http.Request) *livekit.ClientInfo

func (*RTCService) ServeHTTP

func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request)

func (*RTCService) Validate

func (s *RTCService) Validate(w http.ResponseWriter, r *http.Request)

type RedisStore

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

func NewRedisStore

func NewRedisStore(rc redis.UniversalClient) *RedisStore

func (*RedisStore) CleanEndedEgress

func (s *RedisStore) CleanEndedEgress() error

func (*RedisStore) DeleteIngress

func (s *RedisStore) DeleteIngress(_ context.Context, info *livekit.IngressInfo) error

func (*RedisStore) DeleteParticipant

func (s *RedisStore) DeleteParticipant(_ context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity) error

func (*RedisStore) DeleteRoom

func (s *RedisStore) DeleteRoom(ctx context.Context, roomName livekit.RoomName) error

func (*RedisStore) ListEgress

func (s *RedisStore) ListEgress(_ context.Context, roomName livekit.RoomName, active bool) ([]*livekit.EgressInfo, error)

func (*RedisStore) ListIngress

func (s *RedisStore) ListIngress(ctx context.Context, roomName livekit.RoomName) ([]*livekit.IngressInfo, error)

func (*RedisStore) ListParticipants

func (s *RedisStore) ListParticipants(_ context.Context, roomName livekit.RoomName) ([]*livekit.ParticipantInfo, error)

func (*RedisStore) ListRooms

func (s *RedisStore) ListRooms(_ context.Context, roomNames []livekit.RoomName) ([]*livekit.Room, error)

func (*RedisStore) LoadEgress

func (s *RedisStore) LoadEgress(_ context.Context, egressID string) (*livekit.EgressInfo, error)

func (*RedisStore) LoadIngress

func (s *RedisStore) LoadIngress(_ context.Context, ingressId string) (*livekit.IngressInfo, error)

func (*RedisStore) LoadIngressFromStreamKey

func (s *RedisStore) LoadIngressFromStreamKey(_ context.Context, streamKey string) (*livekit.IngressInfo, error)

func (*RedisStore) LoadParticipant

func (s *RedisStore) LoadParticipant(_ context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity) (*livekit.ParticipantInfo, error)

func (*RedisStore) LoadRoom

func (s *RedisStore) LoadRoom(_ context.Context, roomName livekit.RoomName, includeInternal bool) (*livekit.Room, *livekit.RoomInternal, error)

func (*RedisStore) LockRoom

func (s *RedisStore) LockRoom(_ context.Context, roomName livekit.RoomName, duration time.Duration) (string, error)

func (*RedisStore) Start

func (s *RedisStore) Start() error

func (*RedisStore) Stop

func (s *RedisStore) Stop()

func (*RedisStore) StoreEgress

func (s *RedisStore) StoreEgress(_ context.Context, info *livekit.EgressInfo) error

func (*RedisStore) StoreIngress

func (s *RedisStore) StoreIngress(ctx context.Context, info *livekit.IngressInfo) error

func (*RedisStore) StoreParticipant

func (s *RedisStore) StoreParticipant(_ context.Context, roomName livekit.RoomName, participant *livekit.ParticipantInfo) error

func (*RedisStore) StoreRoom

func (s *RedisStore) StoreRoom(_ context.Context, room *livekit.Room, internal *livekit.RoomInternal) error

func (*RedisStore) UnlockRoom

func (s *RedisStore) UnlockRoom(ctx context.Context, roomName livekit.RoomName, uid string) error

func (*RedisStore) UpdateEgress

func (s *RedisStore) UpdateEgress(_ context.Context, info *livekit.EgressInfo) error

func (*RedisStore) UpdateIngress

func (s *RedisStore) UpdateIngress(ctx context.Context, info *livekit.IngressInfo) error

func (*RedisStore) UpdateIngressState

func (s *RedisStore) UpdateIngressState(ctx context.Context, ingressId string, state *livekit.IngressState) error

type RoomAllocator

type RoomAllocator interface {
	CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, error)
	ValidateCreateRoom(ctx context.Context, roomName livekit.RoomName) error
}

func NewRoomAllocator

func NewRoomAllocator(conf *config.Config, router routing.Router, rs ObjectStore) (RoomAllocator, error)

type RoomManager

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

RoomManager manages rooms and its interaction with participants. It's responsible for creating, deleting rooms, as well as running sessions for participants

func NewLocalRoomManager

func NewLocalRoomManager(
	conf *config.Config,
	roomStore ObjectStore,
	currentNode routing.LocalNode,
	router routing.Router,
	telemetry telemetry.TelemetryService,
	clientConfManager clientconfiguration.ClientConfigurationManager,
	egressLauncher rtc.EgressLauncher,
	versionGenerator utils.TimedVersionGenerator,
) (*RoomManager, error)

func (*RoomManager) CleanupRooms

func (r *RoomManager) CleanupRooms() error

CleanupRooms cleans up after old rooms that have been around for a while

func (*RoomManager) CloseIdleRooms

func (r *RoomManager) CloseIdleRooms()

func (*RoomManager) DeleteRoom

func (r *RoomManager) DeleteRoom(ctx context.Context, roomName livekit.RoomName) error

DeleteRoom completely deletes all room information, including active sessions, room store, and routing info

func (*RoomManager) GetRoom

func (r *RoomManager) GetRoom(_ context.Context, roomName livekit.RoomName) *rtc.Room

func (*RoomManager) HasParticipants

func (r *RoomManager) HasParticipants() bool

func (*RoomManager) StartSession

func (r *RoomManager) StartSession(
	ctx context.Context,
	roomName livekit.RoomName,
	pi routing.ParticipantInit,
	requestSource routing.MessageSource,
	responseSink routing.MessageSink,
) error

StartSession starts WebRTC session when a new participant is connected, takes place on RTC node

func (*RoomManager) Stop

func (r *RoomManager) Stop()

type RoomService

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

A rooms service that supports a single node

func NewRoomService

func NewRoomService(
	roomConf config.RoomConfig,
	apiConf config.APIConfig,
	router routing.MessageRouter,
	roomAllocator RoomAllocator,
	serviceStore ServiceStore,
	egressLauncher rtc.EgressLauncher,
) (svc *RoomService, err error)

func (*RoomService) CreateRoom

func (s *RoomService) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, error)

func (*RoomService) DeleteRoom

func (*RoomService) GetParticipant

func (*RoomService) ListParticipants

func (*RoomService) ListRooms

func (*RoomService) MutePublishedTrack

func (*RoomService) RemoveParticipant

func (*RoomService) SendData

func (*RoomService) UpdateParticipant

func (*RoomService) UpdateRoomMetadata

func (s *RoomService) UpdateRoomMetadata(ctx context.Context, req *livekit.UpdateRoomMetadataRequest) (*livekit.Room, error)

type ServiceStore

type ServiceStore interface {
	LoadRoom(ctx context.Context, roomName livekit.RoomName, includeInternal bool) (*livekit.Room, *livekit.RoomInternal, error)

	// ListRooms returns currently active rooms. if names is not nil, it'll filter and return
	// only rooms that match
	ListRooms(ctx context.Context, roomNames []livekit.RoomName) ([]*livekit.Room, error)
	LoadParticipant(ctx context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity) (*livekit.ParticipantInfo, error)
	ListParticipants(ctx context.Context, roomName livekit.RoomName) ([]*livekit.ParticipantInfo, error)
}

type SessionHandler

type SessionHandler func(
	ctx context.Context,
	roomName livekit.RoomName,
	pi routing.ParticipantInit,
	connectionID livekit.ConnectionID,
	requestSource routing.MessageSource,
	responseSink routing.MessageSink,
) error

type SignalServer

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

func NewDefaultSignalServer

func NewDefaultSignalServer(
	currentNode routing.LocalNode,
	bus psrpc.MessageBus,
	config config.SignalRelayConfig,
	router routing.Router,
	roomManager *RoomManager,
) (r *SignalServer, err error)

func NewSignalServer

func NewSignalServer(
	nodeID livekit.NodeID,
	region string,
	bus psrpc.MessageBus,
	config config.SignalRelayConfig,
	sessionHandler SessionHandler,
) (*SignalServer, error)

func (*SignalServer) Stop

func (r *SignalServer) Stop()

type StandardRoomAllocator

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

func (*StandardRoomAllocator) CreateRoom

CreateRoom creates a new room from a request and allocates it to a node to handle it'll also monitor its state, and cleans it up when appropriate

func (*StandardRoomAllocator) ValidateCreateRoom

func (r *StandardRoomAllocator) ValidateCreateRoom(ctx context.Context, roomName livekit.RoomName) error

type WSSignalConnection

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

func NewWSSignalConnection

func NewWSSignalConnection(conn types.WebsocketClient) *WSSignalConnection

func (*WSSignalConnection) ReadRequest

func (c *WSSignalConnection) ReadRequest() (*livekit.SignalRequest, int, error)

func (*WSSignalConnection) WriteResponse

func (c *WSSignalConnection) WriteResponse(msg *livekit.SignalResponse) (int, error)

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