aboutsummaryrefslogtreecommitdiffhomepage
path: root/server/lib/live/live-manager.ts
diff options
context:
space:
mode:
Diffstat (limited to 'server/lib/live/live-manager.ts')
-rw-r--r--server/lib/live/live-manager.ts552
1 files changed, 0 insertions, 552 deletions
diff --git a/server/lib/live/live-manager.ts b/server/lib/live/live-manager.ts
deleted file mode 100644
index acb7af274..000000000
--- a/server/lib/live/live-manager.ts
+++ /dev/null
@@ -1,552 +0,0 @@
1import { readdir, readFile } from 'fs-extra'
2import { createServer, Server } from 'net'
3import { join } from 'path'
4import { createServer as createServerTLS, Server as ServerTLS } from 'tls'
5import { logger, loggerTagsFactory } from '@server/helpers/logger'
6import { CONFIG, registerConfigChangedHandler } from '@server/initializers/config'
7import { VIDEO_LIVE, WEBSERVER } from '@server/initializers/constants'
8import { sequelizeTypescript } from '@server/initializers/database'
9import { RunnerJobModel } from '@server/models/runner/runner-job'
10import { UserModel } from '@server/models/user/user'
11import { VideoModel } from '@server/models/video/video'
12import { VideoLiveModel } from '@server/models/video/video-live'
13import { VideoLiveReplaySettingModel } from '@server/models/video/video-live-replay-setting'
14import { VideoLiveSessionModel } from '@server/models/video/video-live-session'
15import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist'
16import { MVideo, MVideoLiveSession, MVideoLiveVideo, MVideoLiveVideoWithSetting } from '@server/types/models'
17import { pick, wait } from '@shared/core-utils'
18import { ffprobePromise, getVideoStreamBitrate, getVideoStreamDimensionsInfo, getVideoStreamFPS, hasAudioStream } from '@shared/ffmpeg'
19import { LiveVideoError, VideoState } from '@shared/models'
20import { federateVideoIfNeeded } from '../activitypub/videos'
21import { JobQueue } from '../job-queue'
22import { getLiveReplayBaseDirectory } from '../paths'
23import { PeerTubeSocket } from '../peertube-socket'
24import { Hooks } from '../plugins/hooks'
25import { computeResolutionsToTranscode } from '../transcoding/transcoding-resolutions'
26import { LiveQuotaStore } from './live-quota-store'
27import { cleanupAndDestroyPermanentLive, getLiveSegmentTime } from './live-utils'
28import { MuxingSession } from './shared'
29
30const NodeRtmpSession = require('node-media-server/src/node_rtmp_session')
31const context = require('node-media-server/src/node_core_ctx')
32const nodeMediaServerLogger = require('node-media-server/src/node_core_logger')
33
34// Disable node media server logs
35nodeMediaServerLogger.setLogType(0)
36
37const config = {
38 rtmp: {
39 port: CONFIG.LIVE.RTMP.PORT,
40 chunk_size: VIDEO_LIVE.RTMP.CHUNK_SIZE,
41 gop_cache: VIDEO_LIVE.RTMP.GOP_CACHE,
42 ping: VIDEO_LIVE.RTMP.PING,
43 ping_timeout: VIDEO_LIVE.RTMP.PING_TIMEOUT
44 }
45}
46
47const lTags = loggerTagsFactory('live')
48
49class LiveManager {
50
51 private static instance: LiveManager
52
53 private readonly muxingSessions = new Map<string, MuxingSession>()
54 private readonly videoSessions = new Map<string, string>()
55
56 private rtmpServer: Server
57 private rtmpsServer: ServerTLS
58
59 private running = false
60
61 private constructor () {
62 }
63
64 init () {
65 const events = this.getContext().nodeEvent
66 events.on('postPublish', (sessionId: string, streamPath: string) => {
67 logger.debug('RTMP received stream', { id: sessionId, streamPath, ...lTags(sessionId) })
68
69 const splittedPath = streamPath.split('/')
70 if (splittedPath.length !== 3 || splittedPath[1] !== VIDEO_LIVE.RTMP.BASE_PATH) {
71 logger.warn('Live path is incorrect.', { streamPath, ...lTags(sessionId) })
72 return this.abortSession(sessionId)
73 }
74
75 const session = this.getContext().sessions.get(sessionId)
76 const inputLocalUrl = session.inputOriginLocalUrl + streamPath
77 const inputPublicUrl = session.inputOriginPublicUrl + streamPath
78
79 this.handleSession({ sessionId, inputPublicUrl, inputLocalUrl, streamKey: splittedPath[2] })
80 .catch(err => logger.error('Cannot handle sessions.', { err, ...lTags(sessionId) }))
81 })
82
83 events.on('donePublish', sessionId => {
84 logger.info('Live session ended.', { sessionId, ...lTags(sessionId) })
85
86 // Force session aborting, so we kill ffmpeg even if it still has data to process (slow CPU)
87 setTimeout(() => this.abortSession(sessionId), 2000)
88 })
89
90 registerConfigChangedHandler(() => {
91 if (!this.running && CONFIG.LIVE.ENABLED === true) {
92 this.run().catch(err => logger.error('Cannot run live server.', { err }))
93 return
94 }
95
96 if (this.running && CONFIG.LIVE.ENABLED === false) {
97 this.stop()
98 }
99 })
100
101 // Cleanup broken lives, that were terminated by a server restart for example
102 this.handleBrokenLives()
103 .catch(err => logger.error('Cannot handle broken lives.', { err, ...lTags() }))
104 }
105
106 async run () {
107 this.running = true
108
109 if (CONFIG.LIVE.RTMP.ENABLED) {
110 logger.info('Running RTMP server on port %d', CONFIG.LIVE.RTMP.PORT, lTags())
111
112 this.rtmpServer = createServer(socket => {
113 const session = new NodeRtmpSession(config, socket)
114
115 session.inputOriginLocalUrl = 'rtmp://127.0.0.1:' + CONFIG.LIVE.RTMP.PORT
116 session.inputOriginPublicUrl = WEBSERVER.RTMP_URL
117 session.run()
118 })
119
120 this.rtmpServer.on('error', err => {
121 logger.error('Cannot run RTMP server.', { err, ...lTags() })
122 })
123
124 this.rtmpServer.listen(CONFIG.LIVE.RTMP.PORT, CONFIG.LIVE.RTMP.HOSTNAME)
125 }
126
127 if (CONFIG.LIVE.RTMPS.ENABLED) {
128 logger.info('Running RTMPS server on port %d', CONFIG.LIVE.RTMPS.PORT, lTags())
129
130 const [ key, cert ] = await Promise.all([
131 readFile(CONFIG.LIVE.RTMPS.KEY_FILE),
132 readFile(CONFIG.LIVE.RTMPS.CERT_FILE)
133 ])
134 const serverOptions = { key, cert }
135
136 this.rtmpsServer = createServerTLS(serverOptions, socket => {
137 const session = new NodeRtmpSession(config, socket)
138
139 session.inputOriginLocalUrl = 'rtmps://127.0.0.1:' + CONFIG.LIVE.RTMPS.PORT
140 session.inputOriginPublicUrl = WEBSERVER.RTMPS_URL
141 session.run()
142 })
143
144 this.rtmpsServer.on('error', err => {
145 logger.error('Cannot run RTMPS server.', { err, ...lTags() })
146 })
147
148 this.rtmpsServer.listen(CONFIG.LIVE.RTMPS.PORT, CONFIG.LIVE.RTMPS.HOSTNAME)
149 }
150 }
151
152 stop () {
153 this.running = false
154
155 if (this.rtmpServer) {
156 logger.info('Stopping RTMP server.', lTags())
157
158 this.rtmpServer.close()
159 this.rtmpServer = undefined
160 }
161
162 if (this.rtmpsServer) {
163 logger.info('Stopping RTMPS server.', lTags())
164
165 this.rtmpsServer.close()
166 this.rtmpsServer = undefined
167 }
168
169 // Sessions is an object
170 this.getContext().sessions.forEach((session: any) => {
171 if (session instanceof NodeRtmpSession) {
172 session.stop()
173 }
174 })
175 }
176
177 isRunning () {
178 return !!this.rtmpServer
179 }
180
181 hasSession (sessionId: string) {
182 return this.getContext().sessions.has(sessionId)
183 }
184
185 stopSessionOf (videoUUID: string, error: LiveVideoError | null) {
186 const sessionId = this.videoSessions.get(videoUUID)
187 if (!sessionId) {
188 logger.debug('No live session to stop for video %s', videoUUID, lTags(sessionId, videoUUID))
189 return
190 }
191
192 logger.info('Stopping live session of video %s', videoUUID, { error, ...lTags(sessionId, videoUUID) })
193
194 this.saveEndingSession(videoUUID, error)
195 .catch(err => logger.error('Cannot save ending session.', { err, ...lTags(sessionId, videoUUID) }))
196
197 this.videoSessions.delete(videoUUID)
198 this.abortSession(sessionId)
199 }
200
201 private getContext () {
202 return context
203 }
204
205 private abortSession (sessionId: string) {
206 const session = this.getContext().sessions.get(sessionId)
207 if (session) {
208 session.stop()
209 this.getContext().sessions.delete(sessionId)
210 }
211
212 const muxingSession = this.muxingSessions.get(sessionId)
213 if (muxingSession) {
214 // Muxing session will fire and event so we correctly cleanup the session
215 muxingSession.abort()
216
217 this.muxingSessions.delete(sessionId)
218 }
219 }
220
221 private async handleSession (options: {
222 sessionId: string
223 inputLocalUrl: string
224 inputPublicUrl: string
225 streamKey: string
226 }) {
227 const { inputLocalUrl, inputPublicUrl, sessionId, streamKey } = options
228
229 const videoLive = await VideoLiveModel.loadByStreamKey(streamKey)
230 if (!videoLive) {
231 logger.warn('Unknown live video with stream key %s.', streamKey, lTags(sessionId))
232 return this.abortSession(sessionId)
233 }
234
235 const video = videoLive.Video
236 if (video.isBlacklisted()) {
237 logger.warn('Video is blacklisted. Refusing stream %s.', streamKey, lTags(sessionId, video.uuid))
238 return this.abortSession(sessionId)
239 }
240
241 if (this.videoSessions.has(video.uuid)) {
242 logger.warn('Video %s has already a live session. Refusing stream %s.', video.uuid, streamKey, lTags(sessionId, video.uuid))
243 return this.abortSession(sessionId)
244 }
245
246 // Cleanup old potential live (could happen with a permanent live)
247 const oldStreamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id)
248 if (oldStreamingPlaylist) {
249 if (!videoLive.permanentLive) throw new Error('Found previous session in a non permanent live: ' + video.uuid)
250
251 await cleanupAndDestroyPermanentLive(video, oldStreamingPlaylist)
252 }
253
254 this.videoSessions.set(video.uuid, sessionId)
255
256 const now = Date.now()
257 const probe = await ffprobePromise(inputLocalUrl)
258
259 const [ { resolution, ratio }, fps, bitrate, hasAudio ] = await Promise.all([
260 getVideoStreamDimensionsInfo(inputLocalUrl, probe),
261 getVideoStreamFPS(inputLocalUrl, probe),
262 getVideoStreamBitrate(inputLocalUrl, probe),
263 hasAudioStream(inputLocalUrl, probe)
264 ])
265
266 logger.info(
267 '%s probing took %d ms (bitrate: %d, fps: %d, resolution: %d)',
268 inputLocalUrl, Date.now() - now, bitrate, fps, resolution, lTags(sessionId, video.uuid)
269 )
270
271 const allResolutions = await Hooks.wrapObject(
272 this.buildAllResolutionsToTranscode(resolution, hasAudio),
273 'filter:transcoding.auto.resolutions-to-transcode.result',
274 { video }
275 )
276
277 logger.info(
278 'Handling live video of original resolution %d.', resolution,
279 { allResolutions, ...lTags(sessionId, video.uuid) }
280 )
281
282 return this.runMuxingSession({
283 sessionId,
284 videoLive,
285
286 inputLocalUrl,
287 inputPublicUrl,
288 fps,
289 bitrate,
290 ratio,
291 allResolutions,
292 hasAudio
293 })
294 }
295
296 private async runMuxingSession (options: {
297 sessionId: string
298 videoLive: MVideoLiveVideoWithSetting
299
300 inputLocalUrl: string
301 inputPublicUrl: string
302
303 fps: number
304 bitrate: number
305 ratio: number
306 allResolutions: number[]
307 hasAudio: boolean
308 }) {
309 const { sessionId, videoLive } = options
310 const videoUUID = videoLive.Video.uuid
311 const localLTags = lTags(sessionId, videoUUID)
312
313 const liveSession = await this.saveStartingSession(videoLive)
314
315 const user = await UserModel.loadByLiveId(videoLive.id)
316 LiveQuotaStore.Instance.addNewLive(user.id, sessionId)
317
318 const muxingSession = new MuxingSession({
319 context: this.getContext(),
320 sessionId,
321 videoLive,
322 user,
323
324 ...pick(options, [ 'inputLocalUrl', 'inputPublicUrl', 'bitrate', 'ratio', 'fps', 'allResolutions', 'hasAudio' ])
325 })
326
327 muxingSession.on('live-ready', () => this.publishAndFederateLive(videoLive, localLTags))
328
329 muxingSession.on('bad-socket-health', ({ videoUUID }) => {
330 logger.error(
331 'Too much data in client socket stream (ffmpeg is too slow to transcode the video).' +
332 ' Stopping session of video %s.', videoUUID,
333 localLTags
334 )
335
336 this.stopSessionOf(videoUUID, LiveVideoError.BAD_SOCKET_HEALTH)
337 })
338
339 muxingSession.on('duration-exceeded', ({ videoUUID }) => {
340 logger.info('Stopping session of %s: max duration exceeded.', videoUUID, localLTags)
341
342 this.stopSessionOf(videoUUID, LiveVideoError.DURATION_EXCEEDED)
343 })
344
345 muxingSession.on('quota-exceeded', ({ videoUUID }) => {
346 logger.info('Stopping session of %s: user quota exceeded.', videoUUID, localLTags)
347
348 this.stopSessionOf(videoUUID, LiveVideoError.QUOTA_EXCEEDED)
349 })
350
351 muxingSession.on('transcoding-error', ({ videoUUID }) => {
352 this.stopSessionOf(videoUUID, LiveVideoError.FFMPEG_ERROR)
353 })
354
355 muxingSession.on('transcoding-end', ({ videoUUID }) => {
356 this.onMuxingFFmpegEnd(videoUUID, sessionId)
357 })
358
359 muxingSession.on('after-cleanup', ({ videoUUID }) => {
360 this.muxingSessions.delete(sessionId)
361
362 LiveQuotaStore.Instance.removeLive(user.id, sessionId)
363
364 muxingSession.destroy()
365
366 return this.onAfterMuxingCleanup({ videoUUID, liveSession })
367 .catch(err => logger.error('Error in end transmuxing.', { err, ...localLTags }))
368 })
369
370 this.muxingSessions.set(sessionId, muxingSession)
371
372 muxingSession.runMuxing()
373 .catch(err => {
374 logger.error('Cannot run muxing.', { err, ...localLTags })
375 this.abortSession(sessionId)
376 })
377 }
378
379 private async publishAndFederateLive (live: MVideoLiveVideo, localLTags: { tags: string[] }) {
380 const videoId = live.videoId
381
382 try {
383 const video = await VideoModel.loadFull(videoId)
384
385 logger.info('Will publish and federate live %s.', video.url, localLTags)
386
387 video.state = VideoState.PUBLISHED
388 video.publishedAt = new Date()
389 await video.save()
390
391 live.Video = video
392
393 await wait(getLiveSegmentTime(live.latencyMode) * 1000 * VIDEO_LIVE.EDGE_LIVE_DELAY_SEGMENTS_NOTIFICATION)
394
395 try {
396 await federateVideoIfNeeded(video, false)
397 } catch (err) {
398 logger.error('Cannot federate live video %s.', video.url, { err, ...localLTags })
399 }
400
401 PeerTubeSocket.Instance.sendVideoLiveNewState(video)
402
403 Hooks.runAction('action:live.video.state.updated', { video })
404 } catch (err) {
405 logger.error('Cannot save/federate live video %d.', videoId, { err, ...localLTags })
406 }
407 }
408
409 private onMuxingFFmpegEnd (videoUUID: string, sessionId: string) {
410 // Session already cleaned up
411 if (!this.videoSessions.has(videoUUID)) return
412
413 this.videoSessions.delete(videoUUID)
414
415 this.saveEndingSession(videoUUID, null)
416 .catch(err => logger.error('Cannot save ending session.', { err, ...lTags(sessionId) }))
417 }
418
419 private async onAfterMuxingCleanup (options: {
420 videoUUID: string
421 liveSession?: MVideoLiveSession
422 cleanupNow?: boolean // Default false
423 }) {
424 const { videoUUID, liveSession: liveSessionArg, cleanupNow = false } = options
425
426 logger.debug('Live of video %s has been cleaned up. Moving to its next state.', videoUUID, lTags(videoUUID))
427
428 try {
429 const fullVideo = await VideoModel.loadFull(videoUUID)
430 if (!fullVideo) return
431
432 const live = await VideoLiveModel.loadByVideoId(fullVideo.id)
433
434 const liveSession = liveSessionArg ?? await VideoLiveSessionModel.findLatestSessionOf(fullVideo.id)
435
436 // On server restart during a live
437 if (!liveSession.endDate) {
438 liveSession.endDate = new Date()
439 await liveSession.save()
440 }
441
442 JobQueue.Instance.createJobAsync({
443 type: 'video-live-ending',
444 payload: {
445 videoId: fullVideo.id,
446
447 replayDirectory: live.saveReplay
448 ? await this.findReplayDirectory(fullVideo)
449 : undefined,
450
451 liveSessionId: liveSession.id,
452 streamingPlaylistId: fullVideo.getHLSPlaylist()?.id,
453
454 publishedAt: fullVideo.publishedAt.toISOString()
455 },
456
457 delay: cleanupNow
458 ? 0
459 : VIDEO_LIVE.CLEANUP_DELAY
460 })
461
462 fullVideo.state = live.permanentLive
463 ? VideoState.WAITING_FOR_LIVE
464 : VideoState.LIVE_ENDED
465
466 await fullVideo.save()
467
468 PeerTubeSocket.Instance.sendVideoLiveNewState(fullVideo)
469
470 await federateVideoIfNeeded(fullVideo, false)
471
472 Hooks.runAction('action:live.video.state.updated', { video: fullVideo })
473 } catch (err) {
474 logger.error('Cannot save/federate new video state of live streaming of video %s.', videoUUID, { err, ...lTags(videoUUID) })
475 }
476 }
477
478 private async handleBrokenLives () {
479 await RunnerJobModel.cancelAllJobs({ type: 'live-rtmp-hls-transcoding' })
480
481 const videoUUIDs = await VideoModel.listPublishedLiveUUIDs()
482
483 for (const uuid of videoUUIDs) {
484 await this.onAfterMuxingCleanup({ videoUUID: uuid, cleanupNow: true })
485 }
486 }
487
488 private async findReplayDirectory (video: MVideo) {
489 const directory = getLiveReplayBaseDirectory(video)
490 const files = await readdir(directory)
491
492 if (files.length === 0) return undefined
493
494 return join(directory, files.sort().reverse()[0])
495 }
496
497 private buildAllResolutionsToTranscode (originResolution: number, hasAudio: boolean) {
498 const includeInput = CONFIG.LIVE.TRANSCODING.ALWAYS_TRANSCODE_ORIGINAL_RESOLUTION
499
500 const resolutionsEnabled = CONFIG.LIVE.TRANSCODING.ENABLED
501 ? computeResolutionsToTranscode({ input: originResolution, type: 'live', includeInput, strictLower: false, hasAudio })
502 : []
503
504 if (resolutionsEnabled.length === 0) {
505 return [ originResolution ]
506 }
507
508 return resolutionsEnabled
509 }
510
511 private async saveStartingSession (videoLive: MVideoLiveVideoWithSetting) {
512 const replaySettings = videoLive.saveReplay
513 ? new VideoLiveReplaySettingModel({
514 privacy: videoLive.ReplaySetting.privacy
515 })
516 : null
517
518 return sequelizeTypescript.transaction(async t => {
519 if (videoLive.saveReplay) {
520 await replaySettings.save({ transaction: t })
521 }
522
523 return VideoLiveSessionModel.create({
524 startDate: new Date(),
525 liveVideoId: videoLive.videoId,
526 saveReplay: videoLive.saveReplay,
527 replaySettingId: videoLive.saveReplay ? replaySettings.id : null,
528 endingProcessed: false
529 }, { transaction: t })
530 })
531 }
532
533 private async saveEndingSession (videoUUID: string, error: LiveVideoError | null) {
534 const liveSession = await VideoLiveSessionModel.findCurrentSessionOf(videoUUID)
535 if (!liveSession) return
536
537 liveSession.endDate = new Date()
538 liveSession.error = error
539
540 return liveSession.save()
541 }
542
543 static get Instance () {
544 return this.instance || (this.instance = new this())
545 }
546}
547
548// ---------------------------------------------------------------------------
549
550export {
551 LiveManager
552}
generated by cgit v1.2.3 (git 2.25.1) at 2025-04-24 06:02:32 +0000