From 8ebf2a5d5d126e6ef9b89109124adf2a5e9e293d Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Wed, 16 Jun 2021 15:14:41 +0200 Subject: Refactor live manager --- server/lib/live/index.ts | 4 + server/lib/live/live-manager.ts | 412 ++++++++++++++++++++++++++++++ server/lib/live/live-quota-store.ts | 48 ++++ server/lib/live/live-segment-sha-store.ts | 64 +++++ server/lib/live/live-utils.ts | 23 ++ server/lib/live/shared/index.ts | 1 + server/lib/live/shared/muxing-session.ts | 341 +++++++++++++++++++++++++ 7 files changed, 893 insertions(+) create mode 100644 server/lib/live/index.ts create mode 100644 server/lib/live/live-manager.ts create mode 100644 server/lib/live/live-quota-store.ts create mode 100644 server/lib/live/live-segment-sha-store.ts create mode 100644 server/lib/live/live-utils.ts create mode 100644 server/lib/live/shared/index.ts create mode 100644 server/lib/live/shared/muxing-session.ts (limited to 'server/lib/live') diff --git a/server/lib/live/index.ts b/server/lib/live/index.ts new file mode 100644 index 000000000..8b46800da --- /dev/null +++ b/server/lib/live/index.ts @@ -0,0 +1,4 @@ +export * from './live-manager' +export * from './live-quota-store' +export * from './live-segment-sha-store' +export * from './live-utils' diff --git a/server/lib/live/live-manager.ts b/server/lib/live/live-manager.ts new file mode 100644 index 000000000..d7199cc89 --- /dev/null +++ b/server/lib/live/live-manager.ts @@ -0,0 +1,412 @@ + +import { createServer, Server } from 'net' +import { isTestInstance } from '@server/helpers/core-utils' +import { computeResolutionsToTranscode, getVideoFileFPS, getVideoFileResolution } from '@server/helpers/ffprobe-utils' +import { logger, loggerTagsFactory } from '@server/helpers/logger' +import { CONFIG, registerConfigChangedHandler } from '@server/initializers/config' +import { P2P_MEDIA_LOADER_PEER_VERSION, VIDEO_LIVE, VIEW_LIFETIME, WEBSERVER } from '@server/initializers/constants' +import { UserModel } from '@server/models/user/user' +import { VideoModel } from '@server/models/video/video' +import { VideoLiveModel } from '@server/models/video/video-live' +import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist' +import { MStreamingPlaylist, MStreamingPlaylistVideo, MVideo, MVideoLiveVideo } from '@server/types/models' +import { VideoState, VideoStreamingPlaylistType } from '@shared/models' +import { federateVideoIfNeeded } from '../activitypub/videos' +import { JobQueue } from '../job-queue' +import { PeerTubeSocket } from '../peertube-socket' +import { LiveQuotaStore } from './live-quota-store' +import { LiveSegmentShaStore } from './live-segment-sha-store' +import { cleanupLive } from './live-utils' +import { MuxingSession } from './shared' + +const NodeRtmpSession = require('node-media-server/node_rtmp_session') +const context = require('node-media-server/node_core_ctx') +const nodeMediaServerLogger = require('node-media-server/node_core_logger') + +// Disable node media server logs +nodeMediaServerLogger.setLogType(0) + +const config = { + rtmp: { + port: CONFIG.LIVE.RTMP.PORT, + chunk_size: VIDEO_LIVE.RTMP.CHUNK_SIZE, + gop_cache: VIDEO_LIVE.RTMP.GOP_CACHE, + ping: VIDEO_LIVE.RTMP.PING, + ping_timeout: VIDEO_LIVE.RTMP.PING_TIMEOUT + }, + transcoding: { + ffmpeg: 'ffmpeg' + } +} + +const lTags = loggerTagsFactory('live') + +class LiveManager { + + private static instance: LiveManager + + private readonly muxingSessions = new Map() + private readonly videoSessions = new Map() + // Values are Date().getTime() + private readonly watchersPerVideo = new Map() + + private rtmpServer: Server + + private constructor () { + } + + init () { + const events = this.getContext().nodeEvent + events.on('postPublish', (sessionId: string, streamPath: string) => { + logger.debug('RTMP received stream', { id: sessionId, streamPath, ...lTags(sessionId) }) + + const splittedPath = streamPath.split('/') + if (splittedPath.length !== 3 || splittedPath[1] !== VIDEO_LIVE.RTMP.BASE_PATH) { + logger.warn('Live path is incorrect.', { streamPath, ...lTags(sessionId) }) + return this.abortSession(sessionId) + } + + this.handleSession(sessionId, streamPath, splittedPath[2]) + .catch(err => logger.error('Cannot handle sessions.', { err, ...lTags(sessionId) })) + }) + + events.on('donePublish', sessionId => { + logger.info('Live session ended.', { sessionId, ...lTags(sessionId) }) + }) + + registerConfigChangedHandler(() => { + if (!this.rtmpServer && CONFIG.LIVE.ENABLED === true) { + this.run() + return + } + + if (this.rtmpServer && CONFIG.LIVE.ENABLED === false) { + this.stop() + } + }) + + // Cleanup broken lives, that were terminated by a server restart for example + this.handleBrokenLives() + .catch(err => logger.error('Cannot handle broken lives.', { err, ...lTags() })) + + setInterval(() => this.updateLiveViews(), VIEW_LIFETIME.LIVE) + } + + run () { + logger.info('Running RTMP server on port %d', config.rtmp.port, lTags()) + + this.rtmpServer = createServer(socket => { + const session = new NodeRtmpSession(config, socket) + + session.run() + }) + + this.rtmpServer.on('error', err => { + logger.error('Cannot run RTMP server.', { err, ...lTags() }) + }) + + this.rtmpServer.listen(CONFIG.LIVE.RTMP.PORT) + } + + stop () { + logger.info('Stopping RTMP server.', lTags()) + + this.rtmpServer.close() + this.rtmpServer = undefined + + // Sessions is an object + this.getContext().sessions.forEach((session: any) => { + if (session instanceof NodeRtmpSession) { + session.stop() + } + }) + } + + isRunning () { + return !!this.rtmpServer + } + + stopSessionOf (videoId: number) { + const sessionId = this.videoSessions.get(videoId) + if (!sessionId) return + + this.videoSessions.delete(videoId) + this.abortSession(sessionId) + } + + addViewTo (videoId: number) { + if (this.videoSessions.has(videoId) === false) return + + let watchers = this.watchersPerVideo.get(videoId) + + if (!watchers) { + watchers = [] + this.watchersPerVideo.set(videoId, watchers) + } + + watchers.push(new Date().getTime()) + } + + private getContext () { + return context + } + + private abortSession (sessionId: string) { + const session = this.getContext().sessions.get(sessionId) + if (session) { + session.stop() + this.getContext().sessions.delete(sessionId) + } + + const muxingSession = this.muxingSessions.get(sessionId) + if (muxingSession) muxingSession.abort() + } + + private async handleSession (sessionId: string, streamPath: string, streamKey: string) { + const videoLive = await VideoLiveModel.loadByStreamKey(streamKey) + if (!videoLive) { + logger.warn('Unknown live video with stream key %s.', streamKey, lTags(sessionId)) + return this.abortSession(sessionId) + } + + const video = videoLive.Video + if (video.isBlacklisted()) { + logger.warn('Video is blacklisted. Refusing stream %s.', streamKey, lTags(sessionId, video.uuid)) + return this.abortSession(sessionId) + } + + // Cleanup old potential live files (could happen with a permanent live) + LiveSegmentShaStore.Instance.cleanupShaSegments(video.uuid) + + const oldStreamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id) + if (oldStreamingPlaylist) { + await cleanupLive(video, oldStreamingPlaylist) + } + + this.videoSessions.set(video.id, sessionId) + + const rtmpUrl = 'rtmp://127.0.0.1:' + config.rtmp.port + streamPath + + const [ { videoFileResolution }, fps ] = await Promise.all([ + getVideoFileResolution(rtmpUrl), + getVideoFileFPS(rtmpUrl) + ]) + + const allResolutions = this.buildAllResolutionsToTranscode(videoFileResolution) + + logger.info( + 'Will mux/transcode live video of original resolution %d.', videoFileResolution, + { allResolutions, ...lTags(sessionId, video.uuid) } + ) + + const streamingPlaylist = await this.createLivePlaylist(video, allResolutions) + + return this.runMuxingSession({ + sessionId, + videoLive, + streamingPlaylist, + rtmpUrl, + fps, + allResolutions + }) + } + + private async runMuxingSession (options: { + sessionId: string + videoLive: MVideoLiveVideo + streamingPlaylist: MStreamingPlaylistVideo + rtmpUrl: string + fps: number + allResolutions: number[] + }) { + const { sessionId, videoLive, streamingPlaylist, allResolutions, fps, rtmpUrl } = options + const videoUUID = videoLive.Video.uuid + const localLTags = lTags(sessionId, videoUUID) + + const user = await UserModel.loadByLiveId(videoLive.id) + LiveQuotaStore.Instance.addNewLive(user.id, videoLive.id) + + const muxingSession = new MuxingSession({ + context: this.getContext(), + user, + sessionId, + videoLive, + streamingPlaylist, + rtmpUrl, + fps, + allResolutions + }) + + muxingSession.on('master-playlist-created', () => this.publishAndFederateLive(videoLive, localLTags)) + + muxingSession.on('bad-socket-health', ({ videoId }) => { + logger.error( + 'Too much data in client socket stream (ffmpeg is too slow to transcode the video).' + + ' Stopping session of video %s.', videoUUID, + localLTags + ) + + this.stopSessionOf(videoId) + }) + + muxingSession.on('duration-exceeded', ({ videoId }) => { + logger.info('Stopping session of %s: max duration exceeded.', videoUUID, localLTags) + + this.stopSessionOf(videoId) + }) + + muxingSession.on('quota-exceeded', ({ videoId }) => { + logger.info('Stopping session of %s: user quota exceeded.', videoUUID, localLTags) + + this.stopSessionOf(videoId) + }) + + muxingSession.on('ffmpeg-error', ({ sessionId }) => this.abortSession(sessionId)) + muxingSession.on('ffmpeg-end', ({ videoId }) => { + this.onMuxingFFmpegEnd(videoId) + }) + + muxingSession.on('after-cleanup', ({ videoId }) => { + this.muxingSessions.delete(sessionId) + + return this.onAfterMuxingCleanup(videoId) + .catch(err => logger.error('Error in end transmuxing.', { err, ...localLTags })) + }) + + this.muxingSessions.set(sessionId, muxingSession) + + muxingSession.runMuxing() + .catch(err => { + logger.error('Cannot run muxing.', { err, ...localLTags }) + this.abortSession(sessionId) + }) + } + + private async publishAndFederateLive (live: MVideoLiveVideo, localLTags: { tags: string[] }) { + const videoId = live.videoId + + try { + const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoId) + + logger.info('Will publish and federate live %s.', video.url, localLTags) + + video.state = VideoState.PUBLISHED + await video.save() + + live.Video = video + + setTimeout(() => { + federateVideoIfNeeded(video, false) + .catch(err => logger.error('Cannot federate live video %s.', video.url, { err, ...localLTags })) + + PeerTubeSocket.Instance.sendVideoLiveNewState(video) + }, VIDEO_LIVE.SEGMENT_TIME_SECONDS * 1000 * VIDEO_LIVE.EDGE_LIVE_DELAY_SEGMENTS_NOTIFICATION) + } catch (err) { + logger.error('Cannot save/federate live video %d.', videoId, { err, ...localLTags }) + } + } + + private onMuxingFFmpegEnd (videoId: number) { + this.watchersPerVideo.delete(videoId) + this.videoSessions.delete(videoId) + } + + private async onAfterMuxingCleanup (videoUUID: string, cleanupNow = false) { + try { + const fullVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoUUID) + if (!fullVideo) return + + const live = await VideoLiveModel.loadByVideoId(fullVideo.id) + + if (!live.permanentLive) { + JobQueue.Instance.createJob({ + type: 'video-live-ending', + payload: { + videoId: fullVideo.id + } + }, { delay: cleanupNow ? 0 : VIDEO_LIVE.CLEANUP_DELAY }) + + fullVideo.state = VideoState.LIVE_ENDED + } else { + fullVideo.state = VideoState.WAITING_FOR_LIVE + } + + await fullVideo.save() + + PeerTubeSocket.Instance.sendVideoLiveNewState(fullVideo) + + await federateVideoIfNeeded(fullVideo, false) + } catch (err) { + logger.error('Cannot save/federate new video state of live streaming of video %d.', videoUUID, { err, ...lTags(videoUUID) }) + } + } + + private async updateLiveViews () { + if (!this.isRunning()) return + + if (!isTestInstance()) logger.info('Updating live video views.', lTags()) + + for (const videoId of this.watchersPerVideo.keys()) { + const notBefore = new Date().getTime() - VIEW_LIFETIME.LIVE + + const watchers = this.watchersPerVideo.get(videoId) + + const numWatchers = watchers.length + + const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoId) + video.views = numWatchers + await video.save() + + await federateVideoIfNeeded(video, false) + + PeerTubeSocket.Instance.sendVideoViewsUpdate(video) + + // Only keep not expired watchers + const newWatchers = watchers.filter(w => w > notBefore) + this.watchersPerVideo.set(videoId, newWatchers) + + logger.debug('New live video views for %s is %d.', video.url, numWatchers, lTags()) + } + } + + private async handleBrokenLives () { + const videoUUIDs = await VideoModel.listPublishedLiveUUIDs() + + for (const uuid of videoUUIDs) { + await this.onAfterMuxingCleanup(uuid, true) + } + } + + private buildAllResolutionsToTranscode (originResolution: number) { + const resolutionsEnabled = CONFIG.LIVE.TRANSCODING.ENABLED + ? computeResolutionsToTranscode(originResolution, 'live') + : [] + + return resolutionsEnabled.concat([ originResolution ]) + } + + private async createLivePlaylist (video: MVideo, allResolutions: number[]) { + const playlistUrl = WEBSERVER.URL + VideoStreamingPlaylistModel.getHlsMasterPlaylistStaticPath(video.uuid) + const [ videoStreamingPlaylist ] = await VideoStreamingPlaylistModel.upsert({ + videoId: video.id, + playlistUrl, + segmentsSha256Url: WEBSERVER.URL + VideoStreamingPlaylistModel.getHlsSha256SegmentsStaticPath(video.uuid, video.isLive), + p2pMediaLoaderInfohashes: VideoStreamingPlaylistModel.buildP2PMediaLoaderInfoHashes(playlistUrl, allResolutions), + p2pMediaLoaderPeerVersion: P2P_MEDIA_LOADER_PEER_VERSION, + + type: VideoStreamingPlaylistType.HLS + }, { returning: true }) as [ MStreamingPlaylist, boolean ] + + return Object.assign(videoStreamingPlaylist, { Video: video }) + } + + static get Instance () { + return this.instance || (this.instance = new this()) + } +} + +// --------------------------------------------------------------------------- + +export { + LiveManager +} diff --git a/server/lib/live/live-quota-store.ts b/server/lib/live/live-quota-store.ts new file mode 100644 index 000000000..8ceccde98 --- /dev/null +++ b/server/lib/live/live-quota-store.ts @@ -0,0 +1,48 @@ +class LiveQuotaStore { + + private static instance: LiveQuotaStore + + private readonly livesPerUser = new Map() + + private constructor () { + } + + addNewLive (userId: number, liveId: number) { + if (!this.livesPerUser.has(userId)) { + this.livesPerUser.set(userId, []) + } + + const currentUserLive = { liveId, size: 0 } + const livesOfUser = this.livesPerUser.get(userId) + livesOfUser.push(currentUserLive) + } + + removeLive (userId: number, liveId: number) { + const newLivesPerUser = this.livesPerUser.get(userId) + .filter(o => o.liveId !== liveId) + + this.livesPerUser.set(userId, newLivesPerUser) + } + + addQuotaTo (userId: number, liveId: number, size: number) { + const lives = this.livesPerUser.get(userId) + const live = lives.find(l => l.liveId === liveId) + + live.size += size + } + + getLiveQuotaOf (userId: number) { + const currentLives = this.livesPerUser.get(userId) + if (!currentLives) return 0 + + return currentLives.reduce((sum, obj) => sum + obj.size, 0) + } + + static get Instance () { + return this.instance || (this.instance = new this()) + } +} + +export { + LiveQuotaStore +} diff --git a/server/lib/live/live-segment-sha-store.ts b/server/lib/live/live-segment-sha-store.ts new file mode 100644 index 000000000..4af6f3ebf --- /dev/null +++ b/server/lib/live/live-segment-sha-store.ts @@ -0,0 +1,64 @@ +import { basename } from 'path' +import { logger, loggerTagsFactory } from '@server/helpers/logger' +import { buildSha256Segment } from '../hls' + +const lTags = loggerTagsFactory('live') + +class LiveSegmentShaStore { + + private static instance: LiveSegmentShaStore + + private readonly segmentsSha256 = new Map>() + + private constructor () { + } + + getSegmentsSha256 (videoUUID: string) { + return this.segmentsSha256.get(videoUUID) + } + + async addSegmentSha (videoUUID: string, segmentPath: string) { + const segmentName = basename(segmentPath) + logger.debug('Adding live sha segment %s.', segmentPath, lTags(videoUUID)) + + const shaResult = await buildSha256Segment(segmentPath) + + if (!this.segmentsSha256.has(videoUUID)) { + this.segmentsSha256.set(videoUUID, new Map()) + } + + const filesMap = this.segmentsSha256.get(videoUUID) + filesMap.set(segmentName, shaResult) + } + + removeSegmentSha (videoUUID: string, segmentPath: string) { + const segmentName = basename(segmentPath) + + logger.debug('Removing live sha segment %s.', segmentPath, lTags(videoUUID)) + + const filesMap = this.segmentsSha256.get(videoUUID) + if (!filesMap) { + logger.warn('Unknown files map to remove sha for %s.', videoUUID, lTags(videoUUID)) + return + } + + if (!filesMap.has(segmentName)) { + logger.warn('Unknown segment in files map for video %s and segment %s.', videoUUID, segmentPath, lTags(videoUUID)) + return + } + + filesMap.delete(segmentName) + } + + cleanupShaSegments (videoUUID: string) { + this.segmentsSha256.delete(videoUUID) + } + + static get Instance () { + return this.instance || (this.instance = new this()) + } +} + +export { + LiveSegmentShaStore +} diff --git a/server/lib/live/live-utils.ts b/server/lib/live/live-utils.ts new file mode 100644 index 000000000..e4526c7a5 --- /dev/null +++ b/server/lib/live/live-utils.ts @@ -0,0 +1,23 @@ +import { remove } from 'fs-extra' +import { basename } from 'path' +import { MStreamingPlaylist, MVideo } from '@server/types/models' +import { getHLSDirectory } from '../video-paths' + +function buildConcatenatedName (segmentOrPlaylistPath: string) { + const num = basename(segmentOrPlaylistPath).match(/^(\d+)(-|\.)/) + + return 'concat-' + num[1] + '.ts' +} + +async function cleanupLive (video: MVideo, streamingPlaylist: MStreamingPlaylist) { + const hlsDirectory = getHLSDirectory(video) + + await remove(hlsDirectory) + + await streamingPlaylist.destroy() +} + +export { + cleanupLive, + buildConcatenatedName +} diff --git a/server/lib/live/shared/index.ts b/server/lib/live/shared/index.ts new file mode 100644 index 000000000..c4d1b59ec --- /dev/null +++ b/server/lib/live/shared/index.ts @@ -0,0 +1 @@ +export * from './muxing-session' diff --git a/server/lib/live/shared/muxing-session.ts b/server/lib/live/shared/muxing-session.ts new file mode 100644 index 000000000..96f6c2c89 --- /dev/null +++ b/server/lib/live/shared/muxing-session.ts @@ -0,0 +1,341 @@ + +import * as Bluebird from 'bluebird' +import * as chokidar from 'chokidar' +import { FfmpegCommand } from 'fluent-ffmpeg' +import { appendFile, ensureDir, readFile, stat } from 'fs-extra' +import { basename, join } from 'path' +import { EventEmitter } from 'stream' +import { getLiveMuxingCommand, getLiveTranscodingCommand } from '@server/helpers/ffmpeg-utils' +import { logger, loggerTagsFactory, LoggerTagsFn } from '@server/helpers/logger' +import { CONFIG } from '@server/initializers/config' +import { MEMOIZE_TTL, VIDEO_LIVE } from '@server/initializers/constants' +import { VideoFileModel } from '@server/models/video/video-file' +import { MStreamingPlaylistVideo, MUserId, MVideoLiveVideo } from '@server/types/models' +import { VideoTranscodingProfilesManager } from '../../transcoding/video-transcoding-profiles' +import { isAbleToUploadVideo } from '../../user' +import { getHLSDirectory } from '../../video-paths' +import { LiveQuotaStore } from '../live-quota-store' +import { LiveSegmentShaStore } from '../live-segment-sha-store' +import { buildConcatenatedName } from '../live-utils' + +import memoizee = require('memoizee') + +interface MuxingSessionEvents { + 'master-playlist-created': ({ videoId: number }) => void + + 'bad-socket-health': ({ videoId: number }) => void + 'duration-exceeded': ({ videoId: number }) => void + 'quota-exceeded': ({ videoId: number }) => void + + 'ffmpeg-end': ({ videoId: number }) => void + 'ffmpeg-error': ({ sessionId: string }) => void + + 'after-cleanup': ({ videoId: number }) => void +} + +declare interface MuxingSession { + on( + event: U, listener: MuxingSessionEvents[U] + ): this + + emit( + event: U, ...args: Parameters + ): boolean +} + +class MuxingSession extends EventEmitter { + + private ffmpegCommand: FfmpegCommand + + private readonly context: any + private readonly user: MUserId + private readonly sessionId: string + private readonly videoLive: MVideoLiveVideo + private readonly streamingPlaylist: MStreamingPlaylistVideo + private readonly rtmpUrl: string + private readonly fps: number + private readonly allResolutions: number[] + + private readonly videoId: number + private readonly videoUUID: string + private readonly saveReplay: boolean + + private readonly lTags: LoggerTagsFn + + private segmentsToProcessPerPlaylist: { [playlistId: string]: string[] } = {} + + private tsWatcher: chokidar.FSWatcher + private masterWatcher: chokidar.FSWatcher + + private readonly isAbleToUploadVideoWithCache = memoizee((userId: number) => { + return isAbleToUploadVideo(userId, 1000) + }, { maxAge: MEMOIZE_TTL.LIVE_ABLE_TO_UPLOAD }) + + private readonly hasClientSocketInBadHealthWithCache = memoizee((sessionId: string) => { + return this.hasClientSocketInBadHealth(sessionId) + }, { maxAge: MEMOIZE_TTL.LIVE_CHECK_SOCKET_HEALTH }) + + constructor (options: { + context: any + user: MUserId + sessionId: string + videoLive: MVideoLiveVideo + streamingPlaylist: MStreamingPlaylistVideo + rtmpUrl: string + fps: number + allResolutions: number[] + }) { + super() + + this.context = options.context + this.user = options.user + this.sessionId = options.sessionId + this.videoLive = options.videoLive + this.streamingPlaylist = options.streamingPlaylist + this.rtmpUrl = options.rtmpUrl + this.fps = options.fps + this.allResolutions = options.allResolutions + + this.videoId = this.videoLive.Video.id + this.videoUUID = this.videoLive.Video.uuid + + this.saveReplay = this.videoLive.saveReplay + + this.lTags = loggerTagsFactory('live', this.sessionId, this.videoUUID) + } + + async runMuxing () { + this.createFiles() + + const outPath = await this.prepareDirectories() + + this.ffmpegCommand = CONFIG.LIVE.TRANSCODING.ENABLED + ? await getLiveTranscodingCommand({ + rtmpUrl: this.rtmpUrl, + outPath, + resolutions: this.allResolutions, + fps: this.fps, + availableEncoders: VideoTranscodingProfilesManager.Instance.getAvailableEncoders(), + profile: CONFIG.LIVE.TRANSCODING.PROFILE + }) + : getLiveMuxingCommand(this.rtmpUrl, outPath) + + logger.info('Running live muxing/transcoding for %s.', this.videoUUID, this.lTags) + + this.watchTSFiles(outPath) + this.watchMasterFile(outPath) + + this.ffmpegCommand.on('error', (err, stdout, stderr) => { + this.onFFmpegError(err, stdout, stderr, outPath) + }) + + this.ffmpegCommand.on('end', () => this.onFFmpegEnded(outPath)) + + this.ffmpegCommand.run() + } + + abort () { + if (!this.ffmpegCommand) return false + + this.ffmpegCommand.kill('SIGINT') + return true + } + + private onFFmpegError (err: any, stdout: string, stderr: string, outPath: string) { + this.onFFmpegEnded(outPath) + + // Don't care that we killed the ffmpeg process + if (err?.message?.includes('Exiting normally')) return + + logger.error('Live transcoding error.', { err, stdout, stderr, ...this.lTags }) + + this.emit('ffmpeg-error', ({ sessionId: this.sessionId })) + } + + private onFFmpegEnded (outPath: string) { + logger.info('RTMP transmuxing for video %s ended. Scheduling cleanup', this.rtmpUrl, this.lTags) + + setTimeout(() => { + // Wait latest segments generation, and close watchers + + Promise.all([ this.tsWatcher.close(), this.masterWatcher.close() ]) + .then(() => { + // Process remaining segments hash + for (const key of Object.keys(this.segmentsToProcessPerPlaylist)) { + this.processSegments(outPath, this.segmentsToProcessPerPlaylist[key]) + } + }) + .catch(err => { + logger.error( + 'Cannot close watchers of %s or process remaining hash segments.', outPath, + { err, ...this.lTags } + ) + }) + + this.emit('after-cleanup', { videoId: this.videoId }) + }, 1000) + } + + private watchMasterFile (outPath: string) { + this.masterWatcher = chokidar.watch(outPath + '/master.m3u8') + + this.masterWatcher.on('add', async () => { + this.emit('master-playlist-created', { videoId: this.videoId }) + + this.masterWatcher.close() + .catch(err => logger.error('Cannot close master watcher of %s.', outPath, { err, ...this.lTags })) + }) + } + + private watchTSFiles (outPath: string) { + const startStreamDateTime = new Date().getTime() + + this.tsWatcher = chokidar.watch(outPath + '/*.ts') + + const playlistIdMatcher = /^([\d+])-/ + + const addHandler = async segmentPath => { + logger.debug('Live add handler of %s.', segmentPath, this.lTags) + + const playlistId = basename(segmentPath).match(playlistIdMatcher)[0] + + const segmentsToProcess = this.segmentsToProcessPerPlaylist[playlistId] || [] + this.processSegments(outPath, segmentsToProcess) + + this.segmentsToProcessPerPlaylist[playlistId] = [ segmentPath ] + + if (this.hasClientSocketInBadHealthWithCache(this.sessionId)) { + this.emit('bad-socket-health', { videoId: this.videoId }) + return + } + + // Duration constraint check + if (this.isDurationConstraintValid(startStreamDateTime) !== true) { + this.emit('duration-exceeded', { videoId: this.videoId }) + return + } + + // Check user quota if the user enabled replay saving + if (await this.isQuotaExceeded(segmentPath) === true) { + this.emit('quota-exceeded', { videoId: this.videoId }) + } + } + + const deleteHandler = segmentPath => LiveSegmentShaStore.Instance.removeSegmentSha(this.videoUUID, segmentPath) + + this.tsWatcher.on('add', p => addHandler(p)) + this.tsWatcher.on('unlink', p => deleteHandler(p)) + } + + private async isQuotaExceeded (segmentPath: string) { + if (this.saveReplay !== true) return false + + try { + const segmentStat = await stat(segmentPath) + + LiveQuotaStore.Instance.addQuotaTo(this.user.id, this.videoLive.id, segmentStat.size) + + const canUpload = await this.isAbleToUploadVideoWithCache(this.user.id) + + return canUpload !== true + } catch (err) { + logger.error('Cannot stat %s or check quota of %d.', segmentPath, this.user.id, { err, ...this.lTags }) + } + } + + private createFiles () { + for (let i = 0; i < this.allResolutions.length; i++) { + const resolution = this.allResolutions[i] + + const file = new VideoFileModel({ + resolution, + size: -1, + extname: '.ts', + infoHash: null, + fps: this.fps, + videoStreamingPlaylistId: this.streamingPlaylist.id + }) + + VideoFileModel.customUpsert(file, 'streaming-playlist', null) + .catch(err => logger.error('Cannot create file for live streaming.', { err, ...this.lTags })) + } + } + + private async prepareDirectories () { + const outPath = getHLSDirectory(this.videoLive.Video) + await ensureDir(outPath) + + const replayDirectory = join(outPath, VIDEO_LIVE.REPLAY_DIRECTORY) + + if (this.videoLive.saveReplay === true) { + await ensureDir(replayDirectory) + } + + return outPath + } + + private isDurationConstraintValid (streamingStartTime: number) { + const maxDuration = CONFIG.LIVE.MAX_DURATION + // No limit + if (maxDuration < 0) return true + + const now = new Date().getTime() + const max = streamingStartTime + maxDuration + + return now <= max + } + + private processSegments (hlsVideoPath: string, segmentPaths: string[]) { + Bluebird.mapSeries(segmentPaths, async previousSegment => { + // Add sha hash of previous segments, because ffmpeg should have finished generating them + await LiveSegmentShaStore.Instance.addSegmentSha(this.videoUUID, previousSegment) + + if (this.saveReplay) { + await this.addSegmentToReplay(hlsVideoPath, previousSegment) + } + }).catch(err => logger.error('Cannot process segments in %s', hlsVideoPath, { err, ...this.lTags })) + } + + private hasClientSocketInBadHealth (sessionId: string) { + const rtmpSession = this.context.sessions.get(sessionId) + + if (!rtmpSession) { + logger.warn('Cannot get session %s to check players socket health.', sessionId, this.lTags) + return + } + + for (const playerSessionId of rtmpSession.players) { + const playerSession = this.context.sessions.get(playerSessionId) + + if (!playerSession) { + logger.error('Cannot get player session %s to check socket health.', playerSession, this.lTags) + continue + } + + if (playerSession.socket.writableLength > VIDEO_LIVE.MAX_SOCKET_WAITING_DATA) { + return true + } + } + + return false + } + + private async addSegmentToReplay (hlsVideoPath: string, segmentPath: string) { + const segmentName = basename(segmentPath) + const dest = join(hlsVideoPath, VIDEO_LIVE.REPLAY_DIRECTORY, buildConcatenatedName(segmentName)) + + try { + const data = await readFile(segmentPath) + + await appendFile(dest, data) + } catch (err) { + logger.error('Cannot copy segment %s to replay directory.', segmentPath, { err, ...this.lTags }) + } + } +} + +// --------------------------------------------------------------------------- + +export { + MuxingSession +} -- cgit v1.2.3