2 import { createServer, Server } from 'net'
3 import { isTestInstance } from '@server/helpers/core-utils'
4 import { computeResolutionsToTranscode, getVideoFileFPS, getVideoFileResolution } from '@server/helpers/ffprobe-utils'
5 import { logger, loggerTagsFactory } from '@server/helpers/logger'
6 import { CONFIG, registerConfigChangedHandler } from '@server/initializers/config'
7 import { P2P_MEDIA_LOADER_PEER_VERSION, VIDEO_LIVE, VIEW_LIFETIME } from '@server/initializers/constants'
8 import { UserModel } from '@server/models/user/user'
9 import { VideoModel } from '@server/models/video/video'
10 import { VideoLiveModel } from '@server/models/video/video-live'
11 import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist'
12 import { MStreamingPlaylistVideo, MVideo, MVideoLiveVideo } from '@server/types/models'
13 import { VideoState, VideoStreamingPlaylistType } from '@shared/models'
14 import { federateVideoIfNeeded } from '../activitypub/videos'
15 import { JobQueue } from '../job-queue'
16 import { PeerTubeSocket } from '../peertube-socket'
17 import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename } from '../video-paths'
18 import { LiveQuotaStore } from './live-quota-store'
19 import { LiveSegmentShaStore } from './live-segment-sha-store'
20 import { cleanupLive } from './live-utils'
21 import { MuxingSession } from './shared'
23 const NodeRtmpSession = require('node-media-server/src/node_rtmp_session')
24 const context = require('node-media-server/src/node_core_ctx')
25 const nodeMediaServerLogger = require('node-media-server/src/node_core_logger')
27 // Disable node media server logs
28 nodeMediaServerLogger.setLogType(0)
32 port: CONFIG.LIVE.RTMP.PORT,
33 chunk_size: VIDEO_LIVE.RTMP.CHUNK_SIZE,
34 gop_cache: VIDEO_LIVE.RTMP.GOP_CACHE,
35 ping: VIDEO_LIVE.RTMP.PING,
36 ping_timeout: VIDEO_LIVE.RTMP.PING_TIMEOUT
43 const lTags = loggerTagsFactory('live')
47 private static instance: LiveManager
49 private readonly muxingSessions = new Map<string, MuxingSession>()
50 private readonly videoSessions = new Map<number, string>()
51 // Values are Date().getTime()
52 private readonly watchersPerVideo = new Map<number, number[]>()
54 private rtmpServer: Server
56 private constructor () {
60 const events = this.getContext().nodeEvent
61 events.on('postPublish', (sessionId: string, streamPath: string) => {
62 logger.debug('RTMP received stream', { id: sessionId, streamPath, ...lTags(sessionId) })
64 const splittedPath = streamPath.split('/')
65 if (splittedPath.length !== 3 || splittedPath[1] !== VIDEO_LIVE.RTMP.BASE_PATH) {
66 logger.warn('Live path is incorrect.', { streamPath, ...lTags(sessionId) })
67 return this.abortSession(sessionId)
70 this.handleSession(sessionId, streamPath, splittedPath[2])
71 .catch(err => logger.error('Cannot handle sessions.', { err, ...lTags(sessionId) }))
74 events.on('donePublish', sessionId => {
75 logger.info('Live session ended.', { sessionId, ...lTags(sessionId) })
78 registerConfigChangedHandler(() => {
79 if (!this.rtmpServer && CONFIG.LIVE.ENABLED === true) {
84 if (this.rtmpServer && CONFIG.LIVE.ENABLED === false) {
89 // Cleanup broken lives, that were terminated by a server restart for example
90 this.handleBrokenLives()
91 .catch(err => logger.error('Cannot handle broken lives.', { err, ...lTags() }))
93 setInterval(() => this.updateLiveViews(), VIEW_LIFETIME.LIVE)
97 logger.info('Running RTMP server on port %d', config.rtmp.port, lTags())
99 this.rtmpServer = createServer(socket => {
100 const session = new NodeRtmpSession(config, socket)
105 this.rtmpServer.on('error', err => {
106 logger.error('Cannot run RTMP server.', { err, ...lTags() })
109 this.rtmpServer.listen(CONFIG.LIVE.RTMP.PORT)
113 logger.info('Stopping RTMP server.', lTags())
115 this.rtmpServer.close()
116 this.rtmpServer = undefined
118 // Sessions is an object
119 this.getContext().sessions.forEach((session: any) => {
120 if (session instanceof NodeRtmpSession) {
127 return !!this.rtmpServer
130 stopSessionOf (videoId: number) {
131 const sessionId = this.videoSessions.get(videoId)
132 if (!sessionId) return
134 this.videoSessions.delete(videoId)
135 this.abortSession(sessionId)
138 addViewTo (videoId: number) {
139 if (this.videoSessions.has(videoId) === false) return
141 let watchers = this.watchersPerVideo.get(videoId)
145 this.watchersPerVideo.set(videoId, watchers)
148 watchers.push(new Date().getTime())
151 private getContext () {
155 private abortSession (sessionId: string) {
156 const session = this.getContext().sessions.get(sessionId)
159 this.getContext().sessions.delete(sessionId)
162 const muxingSession = this.muxingSessions.get(sessionId)
164 // Muxing session will fire and event so we correctly cleanup the session
165 muxingSession.abort()
167 this.muxingSessions.delete(sessionId)
171 private async handleSession (sessionId: string, streamPath: string, streamKey: string) {
172 const videoLive = await VideoLiveModel.loadByStreamKey(streamKey)
174 logger.warn('Unknown live video with stream key %s.', streamKey, lTags(sessionId))
175 return this.abortSession(sessionId)
178 const video = videoLive.Video
179 if (video.isBlacklisted()) {
180 logger.warn('Video is blacklisted. Refusing stream %s.', streamKey, lTags(sessionId, video.uuid))
181 return this.abortSession(sessionId)
184 // Cleanup old potential live files (could happen with a permanent live)
185 LiveSegmentShaStore.Instance.cleanupShaSegments(video.uuid)
187 const oldStreamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id)
188 if (oldStreamingPlaylist) {
189 await cleanupLive(video, oldStreamingPlaylist)
192 this.videoSessions.set(video.id, sessionId)
194 const rtmpUrl = 'rtmp://127.0.0.1:' + config.rtmp.port + streamPath
196 const [ { videoFileResolution }, fps ] = await Promise.all([
197 getVideoFileResolution(rtmpUrl),
198 getVideoFileFPS(rtmpUrl)
201 const allResolutions = this.buildAllResolutionsToTranscode(videoFileResolution)
204 'Will mux/transcode live video of original resolution %d.', videoFileResolution,
205 { allResolutions, ...lTags(sessionId, video.uuid) }
208 const streamingPlaylist = await this.createLivePlaylist(video, allResolutions)
210 return this.runMuxingSession({
220 private async runMuxingSession (options: {
222 videoLive: MVideoLiveVideo
223 streamingPlaylist: MStreamingPlaylistVideo
226 allResolutions: number[]
228 const { sessionId, videoLive, streamingPlaylist, allResolutions, fps, rtmpUrl } = options
229 const videoUUID = videoLive.Video.uuid
230 const localLTags = lTags(sessionId, videoUUID)
232 const user = await UserModel.loadByLiveId(videoLive.id)
233 LiveQuotaStore.Instance.addNewLive(user.id, videoLive.id)
235 const muxingSession = new MuxingSession({
236 context: this.getContext(),
246 muxingSession.on('master-playlist-created', () => this.publishAndFederateLive(videoLive, localLTags))
248 muxingSession.on('bad-socket-health', ({ videoId }) => {
250 'Too much data in client socket stream (ffmpeg is too slow to transcode the video).' +
251 ' Stopping session of video %s.', videoUUID,
255 this.stopSessionOf(videoId)
258 muxingSession.on('duration-exceeded', ({ videoId }) => {
259 logger.info('Stopping session of %s: max duration exceeded.', videoUUID, localLTags)
261 this.stopSessionOf(videoId)
264 muxingSession.on('quota-exceeded', ({ videoId }) => {
265 logger.info('Stopping session of %s: user quota exceeded.', videoUUID, localLTags)
267 this.stopSessionOf(videoId)
270 muxingSession.on('ffmpeg-error', ({ sessionId }) => this.abortSession(sessionId))
271 muxingSession.on('ffmpeg-end', ({ videoId }) => {
272 this.onMuxingFFmpegEnd(videoId)
275 muxingSession.on('after-cleanup', ({ videoId }) => {
276 this.muxingSessions.delete(sessionId)
278 muxingSession.destroy()
280 return this.onAfterMuxingCleanup(videoId)
281 .catch(err => logger.error('Error in end transmuxing.', { err, ...localLTags }))
284 this.muxingSessions.set(sessionId, muxingSession)
286 muxingSession.runMuxing()
288 logger.error('Cannot run muxing.', { err, ...localLTags })
289 this.abortSession(sessionId)
293 private async publishAndFederateLive (live: MVideoLiveVideo, localLTags: { tags: string[] }) {
294 const videoId = live.videoId
297 const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoId)
299 logger.info('Will publish and federate live %s.', video.url, localLTags)
301 video.state = VideoState.PUBLISHED
307 federateVideoIfNeeded(video, false)
308 .catch(err => logger.error('Cannot federate live video %s.', video.url, { err, ...localLTags }))
310 PeerTubeSocket.Instance.sendVideoLiveNewState(video)
311 }, VIDEO_LIVE.SEGMENT_TIME_SECONDS * 1000 * VIDEO_LIVE.EDGE_LIVE_DELAY_SEGMENTS_NOTIFICATION)
313 logger.error('Cannot save/federate live video %d.', videoId, { err, ...localLTags })
317 private onMuxingFFmpegEnd (videoId: number) {
318 this.watchersPerVideo.delete(videoId)
319 this.videoSessions.delete(videoId)
322 private async onAfterMuxingCleanup (videoUUID: string, cleanupNow = false) {
324 const fullVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoUUID)
325 if (!fullVideo) return
327 const live = await VideoLiveModel.loadByVideoId(fullVideo.id)
329 if (!live.permanentLive) {
330 JobQueue.Instance.createJob({
331 type: 'video-live-ending',
333 videoId: fullVideo.id
335 }, { delay: cleanupNow ? 0 : VIDEO_LIVE.CLEANUP_DELAY })
337 fullVideo.state = VideoState.LIVE_ENDED
339 fullVideo.state = VideoState.WAITING_FOR_LIVE
342 await fullVideo.save()
344 PeerTubeSocket.Instance.sendVideoLiveNewState(fullVideo)
346 await federateVideoIfNeeded(fullVideo, false)
348 logger.error('Cannot save/federate new video state of live streaming of video %d.', videoUUID, { err, ...lTags(videoUUID) })
352 private async updateLiveViews () {
353 if (!this.isRunning()) return
355 if (!isTestInstance()) logger.info('Updating live video views.', lTags())
357 for (const videoId of this.watchersPerVideo.keys()) {
358 const notBefore = new Date().getTime() - VIEW_LIFETIME.LIVE
360 const watchers = this.watchersPerVideo.get(videoId)
362 const numWatchers = watchers.length
364 const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoId)
365 video.views = numWatchers
368 await federateVideoIfNeeded(video, false)
370 PeerTubeSocket.Instance.sendVideoViewsUpdate(video)
372 // Only keep not expired watchers
373 const newWatchers = watchers.filter(w => w > notBefore)
374 this.watchersPerVideo.set(videoId, newWatchers)
376 logger.debug('New live video views for %s is %d.', video.url, numWatchers, lTags())
380 private async handleBrokenLives () {
381 const videoUUIDs = await VideoModel.listPublishedLiveUUIDs()
383 for (const uuid of videoUUIDs) {
384 await this.onAfterMuxingCleanup(uuid, true)
388 private buildAllResolutionsToTranscode (originResolution: number) {
389 const resolutionsEnabled = CONFIG.LIVE.TRANSCODING.ENABLED
390 ? computeResolutionsToTranscode(originResolution, 'live')
393 return resolutionsEnabled.concat([ originResolution ])
396 private async createLivePlaylist (video: MVideo, allResolutions: number[]): Promise<MStreamingPlaylistVideo> {
397 const playlist = await VideoStreamingPlaylistModel.loadOrGenerate(video)
399 playlist.playlistFilename = generateHLSMasterPlaylistFilename(true)
400 playlist.segmentsSha256Filename = generateHlsSha256SegmentsFilename(true)
402 playlist.p2pMediaLoaderPeerVersion = P2P_MEDIA_LOADER_PEER_VERSION
403 playlist.type = VideoStreamingPlaylistType.HLS
405 playlist.assignP2PMediaLoaderInfoHashes(video, allResolutions)
407 return playlist.save()
410 static get Instance () {
411 return this.instance || (this.instance = new this())
415 // ---------------------------------------------------------------------------