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

Merge remote-tracking branch 'origin/master' into raja_async_attributes

Browse Source
This commit is contained in:
boks1971
2026-05-30 12:39:54 +05:30
parent 5ed3a5ae95 7c319a67d4
commit 2295db8ebc
4 changed files with 34 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/participant.go
+7
View File
@@ -411,6 +411,13 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
}
params.Reporter.RegisterFunc(func(ts time.Time, tx roomobs.ParticipantSessionTx) bool {
// Don't publish duration if participant never became active. Otherwise short-lived
// JOINING/JOINED -> DISCONNECTED transitions would still get rounded up to a
// minute by the session timer and inflate billed/reported duration.
if p.lastActiveAt.Load() == nil {
return !p.IsClosed()
}
if dts := p.disconnectedAt.Load(); dts != nil {
ts = *dts
tx.ReportEndTime(ts)
pkg/service/wire.go
+9 -1
View File
@@ -60,7 +60,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
wire.Bind(new(routing.MessageRouter), new(routing.Router)),
wire.Bind(new(livekit.RoomService), new(*RoomService)),
telemetry.NewAnalyticsService,
telemetry.NewTelemetryService,
createTelemetryService,
getMessageBus,
NewIOInfoService,
wire.Bind(new(IOClient), new(*IOInfoService)),
@@ -171,6 +171,14 @@ func createWebhookNotifier(conf *config.Config, provider auth.KeyProvider) (webh
return webhook.NewDefaultNotifier(wc, provider)
}
func createTelemetryService(notifier webhook.QueuedNotifier, analytics telemetry.AnalyticsService) telemetry.TelemetryService {
svc := telemetry.NewTelemetryService(notifier, analytics)
if notifier != nil {
notifier.RegisterProcessedHook(svc.Webhook)
}
return svc
}
func createRedisClient(conf *config.Config) (redis.UniversalClient, error) {
if !conf.Redis.IsConfigured() {
return nil, nil
pkg/service/wire_gen.go
+18 -10
View File
@@ -83,30 +83,30 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
return nil, err
}
analyticsService := telemetry.NewAnalyticsService(conf, currentNode)
telemetryService := telemetry.NewTelemetryService(queuedNotifier, analyticsService)
telemetryService := createTelemetryService(queuedNotifier, analyticsService)
ioInfoService, err := NewIOInfoService(messageBus, egressStore, ingressStore, sipStore, telemetryService)
if err != nil {
return nil, err
}
rtcEgressLauncher := NewEgressLauncher(egressClient, ioInfoService, objectStore)
topicFormatter := rpc.NewTopicFormatter()
roomClient, err := rpc.NewTypedRoomClient(clientParams)
v, err := rpc.NewTypedRoomClient(clientParams)
if err != nil {
return nil, err
}
participantClient, err := rpc.NewTypedParticipantClient(clientParams)
v2, err := rpc.NewTypedParticipantClient(clientParams)
if err != nil {
return nil, err
}
roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, roomClient, participantClient)
roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, v, v2)
if err != nil {
return nil, err
}
agentDispatchInternalClient, err := rpc.NewTypedAgentDispatchInternalClient(clientParams)
v3, err := rpc.NewTypedAgentDispatchInternalClient(clientParams)
if err != nil {
return nil, err
}
agentDispatchService := NewAgentDispatchService(agentDispatchInternalClient, topicFormatter, roomAllocator, router)
agentDispatchService := NewAgentDispatchService(v3, topicFormatter, roomAllocator, router)
egressService := NewEgressService(egressClient, rtcEgressLauncher, ioInfoService, roomService)
ingressConfig := getIngressConfig(conf)
ingressClient, err := rpc.NewIngressClient(clientParams)
@@ -121,11 +121,11 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
}
sipService := NewSIPService(sipConfig, nodeID, messageBus, sipClient, sipStore, roomService, telemetryService)
rtcService := NewRTCService(conf, roomAllocator, router, telemetryService)
whipParticipantClient, err := rpc.NewTypedWHIPParticipantClient(clientParams)
v4, err := rpc.NewTypedWHIPParticipantClient(clientParams)
if err != nil {
return nil, err
}
serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, whipParticipantClient)
serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, v4)
if err != nil {
return nil, err
}
@@ -150,8 +150,8 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
if err != nil {
return nil, err
}
authHandler := getTURNAuthHandlerFunc(turnAuthHandler)
server, err := newInProcessTurnServer(conf, authHandler)
v5 := getTURNAuthHandlerFunc(turnAuthHandler)
server, err := newInProcessTurnServer(conf, v5)
if err != nil {
return nil, err
}
@@ -236,6 +236,14 @@ func createWebhookNotifier(conf *config.Config, provider auth.KeyProvider) (webh
return webhook.NewDefaultNotifier(wc, provider)
}
func createTelemetryService(notifier webhook.QueuedNotifier, analytics telemetry.AnalyticsService) telemetry.TelemetryService {
svc := telemetry.NewTelemetryService(notifier, analytics)
if notifier != nil {
notifier.RegisterProcessedHook(svc.Webhook)
}
return svc
}
func createRedisClient(conf *config.Config) (redis.UniversalClient, error) {
if !conf.Redis.IsConfigured() {
return nil, nil
pkg/telemetry/telemetryservice.go
-5
View File
@@ -195,11 +195,6 @@ func NewTelemetryService(notifier webhook.QueuedNotifier, analytics AnalyticsSer
}),
workers: make(map[statsWorkerKey]*StatsWorker),
}
if t.notifier != nil {
t.notifier.RegisterProcessedHook(func(ctx context.Context, whi *livekit.WebhookInfo) {
t.Webhook(ctx, whi)
})
}
t.jobsQueue.Start()
go t.run()