Refactor live manager

7 files changed, 893 insertions, 0 deletions

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<string, MuxingSession>()
+  private readonly videoSessions = new Map<number, string>()
+  // Values are Date().getTime()
+  private readonly watchersPerVideo = new Map<number, number[]>()
+
+  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<number, { liveId: number, size: number }[]>()
+
+  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<string, Map<string, string>>()
+
+  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<U extends keyof MuxingSessionEvents>(
+    event: U, listener: MuxingSessionEvents[U]
+  ): this
+
+  emit<U extends keyof MuxingSessionEvents>(
+    event: U, ...args: Parameters<MuxingSessionEvents[U]>
+  ): 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
+}