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

Add support for Ingress service (#868)

Browse Source 
Also check for Join permission on ingress API calls
This commit is contained in:
Benjamin Pracht
2022-08-04 09:48:53 -07:00
committed by GitHub
parent 6b6f61b4df
commit c333b7a026
5 changed files with 80 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config-sample.yaml
+7
View File
@@ -176,6 +176,13 @@ keys:
# # cert_file: /path/to/cert.pem
# # key_file: /path/to/key.pem
# ingress server
# ingress:
# # Prefix used to generate RTMP URLs for RTMP ingress.
# # The stream_key will be appended to this base and returned as part of the
# # ingress info
# rtmp_base_url: "rtmp://my.domain.com/live"
# Region of the current node. Required if using regionaware node selector
# region: us-west-2
pkg/service/ingress.go
+21 -6
View File
@@ -25,7 +25,7 @@ type IngressService struct {
}
func NewIngressService(
conf *config.IngressConfig,
conf *config.Config,
rpc ingress.RPC,
store IngressStore,
rs livekit.RoomService,
@@ -33,7 +33,7 @@ func NewIngressService(
) *IngressService {
return &IngressService{
conf: conf,
conf: &conf.Ingress,
rpc: rpc,
store: store,
roomService: rs,
@@ -54,7 +54,11 @@ func (s *IngressService) Stop() {
}
func (s *IngressService) CreateIngress(ctx context.Context, req *livekit.CreateIngressRequest) (*livekit.IngressInfo, error) {
if err := EnsureRecordPermission(ctx); err != nil {
roomName, err := EnsureJoinPermission(ctx)
if err != nil {
return nil, twirpAuthError(err)
}
if req.RoomName != "" && req.RoomName != string(roomName) {
return nil, twirpAuthError(err)
}
@@ -86,9 +90,14 @@ func (s *IngressService) CreateIngress(ctx context.Context, req *livekit.CreateI
}
func (s *IngressService) UpdateIngress(ctx context.Context, req *livekit.UpdateIngressRequest) (*livekit.IngressInfo, error) {
if err := EnsureRecordPermission(ctx); err != nil {
roomName, err := EnsureJoinPermission(ctx)
if err != nil {
return nil, twirpAuthError(err)
}
if req.RoomName != "" && req.RoomName != string(roomName) {
return nil, twirpAuthError(err)
}
if s.rpc == nil {
return nil, ErrIngressNotConnected
}
@@ -146,9 +155,14 @@ func (s *IngressService) UpdateIngress(ctx context.Context, req *livekit.UpdateI
}
func (s *IngressService) ListIngress(ctx context.Context, req *livekit.ListIngressRequest) (*livekit.ListIngressResponse, error) {
if err := EnsureRecordPermission(ctx); err != nil {
roomName, err := EnsureJoinPermission(ctx)
if err != nil {
return nil, twirpAuthError(err)
}
if req.RoomName != "" && req.RoomName != string(roomName) {
return nil, twirpAuthError(err)
}
if s.rpc == nil {
return nil, ErrIngressNotConnected
}
@@ -163,9 +177,10 @@ func (s *IngressService) ListIngress(ctx context.Context, req *livekit.ListIngre
}
func (s *IngressService) DeleteIngress(ctx context.Context, req *livekit.DeleteIngressRequest) (*livekit.IngressInfo, error) {
if err := EnsureRecordPermission(ctx); err != nil {
if _, err := EnsureJoinPermission(ctx); err != nil {
return nil, twirpAuthError(err)
}
if s.rpc == nil {
return nil, ErrIngressNotConnected
}
pkg/service/server.go
+25 -17
View File
@@ -27,23 +27,25 @@ import (
)
type LivekitServer struct {
config *config.Config
egressService *EgressService
rtcService *RTCService
httpServer *http.Server
promServer *http.Server
router routing.Router
roomManager *RoomManager
turnServer *turn.Server
currentNode routing.LocalNode
running atomic.Bool
doneChan chan struct{}
closedChan chan struct{}
config *config.Config
egressService *EgressService
ingressService *IngressService
rtcService *RTCService
httpServer *http.Server
promServer *http.Server
router routing.Router
roomManager *RoomManager
turnServer *turn.Server
currentNode routing.LocalNode
running atomic.Bool
doneChan chan struct{}
closedChan chan struct{}
}
func NewLivekitServer(conf *config.Config,
roomService livekit.RoomService,
egressService *EgressService,
ingressService *IngressService,
rtcService *RTCService,
keyProvider auth.KeyProvider,
router routing.Router,
@@ -52,11 +54,12 @@ func NewLivekitServer(conf *config.Config,
currentNode routing.LocalNode,
) (s *LivekitServer, err error) {
s = &LivekitServer{
config: conf,
egressService: egressService,
rtcService: rtcService,
router: router,
roomManager: roomManager,
config: conf,
egressService: egressService,
ingressService: ingressService,
rtcService: rtcService,
router: router,
roomManager: roomManager,
// turn server starts automatically
turnServer: turnServer,
currentNode: currentNode,
@@ -80,6 +83,7 @@ func NewLivekitServer(conf *config.Config,
roomServer := livekit.NewRoomServiceServer(roomService)
egressServer := livekit.NewEgressServer(egressService)
ingressServer := livekit.NewIngressServer(ingressService)
mux := http.NewServeMux()
if conf.Development {
@@ -90,6 +94,7 @@ func NewLivekitServer(conf *config.Config,
}
mux.Handle(roomServer.PathPrefix(), roomServer)
mux.Handle(egressServer.PathPrefix(), egressServer)
mux.Handle(ingressServer.PathPrefix(), ingressServer)
mux.Handle("/rtc", rtcService)
mux.HandleFunc("/rtc/validate", rtcService.Validate)
mux.HandleFunc("/", s.healthCheck)
@@ -150,6 +155,8 @@ func (s *LivekitServer) Start() error {
return err
}
s.ingressService.Start()
addresses := s.config.BindAddresses
if addresses == nil {
addresses = []string{""}
@@ -238,6 +245,7 @@ func (s *LivekitServer) Start() error {
s.roomManager.Stop()
s.egressService.Stop()
s.ingressService.Stop()
close(s.closedChan)
return nil
pkg/service/wire.go
+13
View File
@@ -16,6 +16,7 @@ import (
"github.com/livekit/protocol/auth"
"github.com/livekit/protocol/egress"
"github.com/livekit/protocol/ingress"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/webhook"
@@ -44,6 +45,9 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
egress.NewRedisRPCClient,
getEgressStore,
NewEgressService,
ingress.NewRedisRPC,
getIngressStore,
NewIngressService,
NewRoomAllocator,
NewRoomService,
NewRTCService,
@@ -175,6 +179,15 @@ func getEgressStore(s ObjectStore) EgressStore {
}
}
func getIngressStore(s ObjectStore) IngressStore {
switch store := s.(type) {
case *RedisStore:
return store
default:
return nil
}
}
func createClientConfiguration() clientconfiguration.ClientConfigurationManager {
return clientconfiguration.NewStaticClientConfigurationManager(clientconfiguration.StaticConfigurations)
}
pkg/service/wire_gen.go
+14 -1
View File
@@ -17,6 +17,7 @@ import (
"github.com/livekit/livekit-server/pkg/telemetry"
"github.com/livekit/protocol/auth"
"github.com/livekit/protocol/egress"
"github.com/livekit/protocol/ingress"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/webhook"
@@ -61,6 +62,9 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
analyticsService := telemetry.NewAnalyticsService(conf, currentNode)
telemetryService := telemetry.NewTelemetryService(notifier, analyticsService)
egressService := NewEgressService(rpcClient, objectStore, egressStore, roomService, telemetryService)
rpc := ingress.NewRedisRPC(nodeID, client)
ingressStore := getIngressStore(objectStore)
ingressService := NewIngressService(conf, rpc, ingressStore, roomService, telemetryService)
rtcService := NewRTCService(conf, roomAllocator, objectStore, router, currentNode)
clientConfigurationManager := createClientConfiguration()
roomManager, err := NewLocalRoomManager(conf, objectStore, currentNode, router, telemetryService, clientConfigurationManager)
@@ -72,7 +76,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
if err != nil {
return nil, err
}
livekitServer, err := NewLivekitServer(conf, roomService, egressService, rtcService, keyProvider, router, roomManager, server, currentNode)
livekitServer, err := NewLivekitServer(conf, roomService, egressService, ingressService, rtcService, keyProvider, router, roomManager, server, currentNode)
if err != nil {
return nil, err
}
@@ -201,6 +205,15 @@ func getEgressStore(s ObjectStore) EgressStore {
}
}
func getIngressStore(s ObjectStore) IngressStore {
switch store := s.(type) {
case *RedisStore:
return store
default:
return nil
}
}
func createClientConfiguration() clientconfiguration.ClientConfigurationManager {
return clientconfiguration.NewStaticClientConfigurationManager(clientconfiguration.StaticConfigurations)
}