]>
Commit | Line | Data |
---|---|---|
8ebf2a5d | 1 | |
41fb13c3 C |
2 | import { mapSeries } from 'bluebird' |
3 | import { FSWatcher, watch } from 'chokidar' | |
8ebf2a5d C |
4 | import { FfmpegCommand } from 'fluent-ffmpeg' |
5 | import { appendFile, ensureDir, readFile, stat } from 'fs-extra' | |
6 | import { basename, join } from 'path' | |
7 | import { EventEmitter } from 'stream' | |
c729caf6 | 8 | import { getLiveMuxingCommand, getLiveTranscodingCommand } from '@server/helpers/ffmpeg' |
8ebf2a5d C |
9 | import { logger, loggerTagsFactory, LoggerTagsFn } from '@server/helpers/logger' |
10 | import { CONFIG } from '@server/initializers/config' | |
11 | import { MEMOIZE_TTL, VIDEO_LIVE } from '@server/initializers/constants' | |
12 | import { VideoFileModel } from '@server/models/video/video-file' | |
13 | import { MStreamingPlaylistVideo, MUserId, MVideoLiveVideo } from '@server/types/models' | |
4ec52d04 | 14 | import { getLiveDirectory, getLiveReplayBaseDirectory } from '../../paths' |
c729caf6 | 15 | import { VideoTranscodingProfilesManager } from '../../transcoding/default-transcoding-profiles' |
8ebf2a5d | 16 | import { isAbleToUploadVideo } from '../../user' |
8ebf2a5d C |
17 | import { LiveQuotaStore } from '../live-quota-store' |
18 | import { LiveSegmentShaStore } from '../live-segment-sha-store' | |
19 | import { buildConcatenatedName } from '../live-utils' | |
20 | ||
21 | import memoizee = require('memoizee') | |
22 | ||
23 | interface MuxingSessionEvents { | |
15eb9e5b | 24 | 'master-playlist-created': (options: { videoId: number }) => void |
8ebf2a5d | 25 | |
15eb9e5b C |
26 | 'bad-socket-health': (options: { videoId: number }) => void |
27 | 'duration-exceeded': (options: { videoId: number }) => void | |
28 | 'quota-exceeded': (options: { videoId: number }) => void | |
8ebf2a5d | 29 | |
15eb9e5b C |
30 | 'ffmpeg-end': (options: { videoId: number }) => void |
31 | 'ffmpeg-error': (options: { videoId: number }) => void | |
8ebf2a5d | 32 | |
15eb9e5b | 33 | 'after-cleanup': (options: { videoId: number }) => void |
8ebf2a5d C |
34 | } |
35 | ||
36 | declare interface MuxingSession { | |
37 | on<U extends keyof MuxingSessionEvents>( | |
38 | event: U, listener: MuxingSessionEvents[U] | |
39 | ): this | |
40 | ||
41 | emit<U extends keyof MuxingSessionEvents>( | |
42 | event: U, ...args: Parameters<MuxingSessionEvents[U]> | |
43 | ): boolean | |
44 | } | |
45 | ||
46 | class MuxingSession extends EventEmitter { | |
47 | ||
48 | private ffmpegCommand: FfmpegCommand | |
49 | ||
50 | private readonly context: any | |
51 | private readonly user: MUserId | |
52 | private readonly sessionId: string | |
53 | private readonly videoLive: MVideoLiveVideo | |
54 | private readonly streamingPlaylist: MStreamingPlaylistVideo | |
df1db951 | 55 | private readonly inputUrl: string |
8ebf2a5d C |
56 | private readonly fps: number |
57 | private readonly allResolutions: number[] | |
58 | ||
679c12e6 C |
59 | private readonly bitrate: number |
60 | private readonly ratio: number | |
61 | ||
1ce4256a C |
62 | private readonly hasAudio: boolean |
63 | ||
8ebf2a5d C |
64 | private readonly videoId: number |
65 | private readonly videoUUID: string | |
66 | private readonly saveReplay: boolean | |
67 | ||
4ec52d04 C |
68 | private readonly outDirectory: string |
69 | private readonly replayDirectory: string | |
70 | ||
8ebf2a5d C |
71 | private readonly lTags: LoggerTagsFn |
72 | ||
73 | private segmentsToProcessPerPlaylist: { [playlistId: string]: string[] } = {} | |
74 | ||
41fb13c3 C |
75 | private tsWatcher: FSWatcher |
76 | private masterWatcher: FSWatcher | |
8ebf2a5d | 77 | |
39d117a4 C |
78 | private aborted = false |
79 | ||
8ebf2a5d C |
80 | private readonly isAbleToUploadVideoWithCache = memoizee((userId: number) => { |
81 | return isAbleToUploadVideo(userId, 1000) | |
82 | }, { maxAge: MEMOIZE_TTL.LIVE_ABLE_TO_UPLOAD }) | |
83 | ||
84 | private readonly hasClientSocketInBadHealthWithCache = memoizee((sessionId: string) => { | |
85 | return this.hasClientSocketInBadHealth(sessionId) | |
86 | }, { maxAge: MEMOIZE_TTL.LIVE_CHECK_SOCKET_HEALTH }) | |
87 | ||
88 | constructor (options: { | |
89 | context: any | |
90 | user: MUserId | |
91 | sessionId: string | |
92 | videoLive: MVideoLiveVideo | |
93 | streamingPlaylist: MStreamingPlaylistVideo | |
df1db951 | 94 | inputUrl: string |
8ebf2a5d | 95 | fps: number |
c826f34a | 96 | bitrate: number |
679c12e6 | 97 | ratio: number |
8ebf2a5d | 98 | allResolutions: number[] |
1ce4256a | 99 | hasAudio: boolean |
8ebf2a5d C |
100 | }) { |
101 | super() | |
102 | ||
103 | this.context = options.context | |
104 | this.user = options.user | |
105 | this.sessionId = options.sessionId | |
106 | this.videoLive = options.videoLive | |
107 | this.streamingPlaylist = options.streamingPlaylist | |
df1db951 | 108 | this.inputUrl = options.inputUrl |
8ebf2a5d | 109 | this.fps = options.fps |
679c12e6 | 110 | |
c826f34a | 111 | this.bitrate = options.bitrate |
41085b15 | 112 | this.ratio = options.ratio |
679c12e6 | 113 | |
1ce4256a C |
114 | this.hasAudio = options.hasAudio |
115 | ||
8ebf2a5d C |
116 | this.allResolutions = options.allResolutions |
117 | ||
118 | this.videoId = this.videoLive.Video.id | |
119 | this.videoUUID = this.videoLive.Video.uuid | |
120 | ||
121 | this.saveReplay = this.videoLive.saveReplay | |
122 | ||
4ec52d04 C |
123 | this.outDirectory = getLiveDirectory(this.videoLive.Video) |
124 | this.replayDirectory = join(getLiveReplayBaseDirectory(this.videoLive.Video), new Date().toISOString()) | |
125 | ||
8ebf2a5d C |
126 | this.lTags = loggerTagsFactory('live', this.sessionId, this.videoUUID) |
127 | } | |
128 | ||
129 | async runMuxing () { | |
130 | this.createFiles() | |
131 | ||
4ec52d04 | 132 | await this.prepareDirectories() |
8ebf2a5d C |
133 | |
134 | this.ffmpegCommand = CONFIG.LIVE.TRANSCODING.ENABLED | |
135 | ? await getLiveTranscodingCommand({ | |
df1db951 | 136 | inputUrl: this.inputUrl, |
764b1a14 | 137 | |
4ec52d04 | 138 | outPath: this.outDirectory, |
764b1a14 C |
139 | masterPlaylistName: this.streamingPlaylist.playlistFilename, |
140 | ||
f443a746 C |
141 | latencyMode: this.videoLive.latencyMode, |
142 | ||
8ebf2a5d C |
143 | resolutions: this.allResolutions, |
144 | fps: this.fps, | |
c826f34a | 145 | bitrate: this.bitrate, |
679c12e6 | 146 | ratio: this.ratio, |
c826f34a | 147 | |
1ce4256a C |
148 | hasAudio: this.hasAudio, |
149 | ||
8ebf2a5d C |
150 | availableEncoders: VideoTranscodingProfilesManager.Instance.getAvailableEncoders(), |
151 | profile: CONFIG.LIVE.TRANSCODING.PROFILE | |
152 | }) | |
f443a746 C |
153 | : getLiveMuxingCommand({ |
154 | inputUrl: this.inputUrl, | |
4ec52d04 | 155 | outPath: this.outDirectory, |
f443a746 C |
156 | masterPlaylistName: this.streamingPlaylist.playlistFilename, |
157 | latencyMode: this.videoLive.latencyMode | |
158 | }) | |
8ebf2a5d | 159 | |
4c6757f2 | 160 | logger.info('Running live muxing/transcoding for %s.', this.videoUUID, this.lTags()) |
8ebf2a5d | 161 | |
b34ee7fa C |
162 | this.watchTSFiles() |
163 | this.watchMasterFile() | |
8ebf2a5d | 164 | |
4c6757f2 C |
165 | let ffmpegShellCommand: string |
166 | this.ffmpegCommand.on('start', cmdline => { | |
167 | ffmpegShellCommand = cmdline | |
168 | ||
169 | logger.debug('Running ffmpeg command for live', { ffmpegShellCommand, ...this.lTags() }) | |
170 | }) | |
171 | ||
8ebf2a5d | 172 | this.ffmpegCommand.on('error', (err, stdout, stderr) => { |
b34ee7fa | 173 | this.onFFmpegError({ err, stdout, stderr, ffmpegShellCommand }) |
8ebf2a5d C |
174 | }) |
175 | ||
26e3e98f C |
176 | this.ffmpegCommand.on('end', () => { |
177 | this.emit('ffmpeg-end', ({ videoId: this.videoId })) | |
178 | ||
b34ee7fa | 179 | this.onFFmpegEnded() |
26e3e98f | 180 | }) |
8ebf2a5d C |
181 | |
182 | this.ffmpegCommand.run() | |
183 | } | |
184 | ||
185 | abort () { | |
609a4442 | 186 | if (!this.ffmpegCommand) return |
8ebf2a5d | 187 | |
39d117a4 | 188 | this.aborted = true |
8ebf2a5d | 189 | this.ffmpegCommand.kill('SIGINT') |
609a4442 C |
190 | } |
191 | ||
192 | destroy () { | |
193 | this.removeAllListeners() | |
194 | this.isAbleToUploadVideoWithCache.clear() | |
195 | this.hasClientSocketInBadHealthWithCache.clear() | |
8ebf2a5d C |
196 | } |
197 | ||
4c6757f2 C |
198 | private onFFmpegError (options: { |
199 | err: any | |
200 | stdout: string | |
201 | stderr: string | |
4c6757f2 C |
202 | ffmpegShellCommand: string |
203 | }) { | |
b34ee7fa | 204 | const { err, stdout, stderr, ffmpegShellCommand } = options |
4c6757f2 | 205 | |
b34ee7fa | 206 | this.onFFmpegEnded() |
8ebf2a5d C |
207 | |
208 | // Don't care that we killed the ffmpeg process | |
209 | if (err?.message?.includes('Exiting normally')) return | |
210 | ||
4c6757f2 | 211 | logger.error('Live transcoding error.', { err, stdout, stderr, ffmpegShellCommand, ...this.lTags() }) |
8ebf2a5d | 212 | |
26e3e98f | 213 | this.emit('ffmpeg-error', ({ videoId: this.videoId })) |
8ebf2a5d C |
214 | } |
215 | ||
b34ee7fa | 216 | private onFFmpegEnded () { |
4c6757f2 | 217 | logger.info('RTMP transmuxing for video %s ended. Scheduling cleanup', this.inputUrl, this.lTags()) |
8ebf2a5d C |
218 | |
219 | setTimeout(() => { | |
220 | // Wait latest segments generation, and close watchers | |
221 | ||
222 | Promise.all([ this.tsWatcher.close(), this.masterWatcher.close() ]) | |
223 | .then(() => { | |
224 | // Process remaining segments hash | |
225 | for (const key of Object.keys(this.segmentsToProcessPerPlaylist)) { | |
b34ee7fa | 226 | this.processSegments(this.segmentsToProcessPerPlaylist[key]) |
8ebf2a5d C |
227 | } |
228 | }) | |
229 | .catch(err => { | |
230 | logger.error( | |
b34ee7fa | 231 | 'Cannot close watchers of %s or process remaining hash segments.', this.outDirectory, |
4c6757f2 | 232 | { err, ...this.lTags() } |
8ebf2a5d C |
233 | ) |
234 | }) | |
235 | ||
236 | this.emit('after-cleanup', { videoId: this.videoId }) | |
237 | }, 1000) | |
238 | } | |
239 | ||
b34ee7fa C |
240 | private watchMasterFile () { |
241 | this.masterWatcher = watch(this.outDirectory + '/' + this.streamingPlaylist.playlistFilename) | |
8ebf2a5d | 242 | |
98ab5dc8 | 243 | this.masterWatcher.on('add', () => { |
8ebf2a5d C |
244 | this.emit('master-playlist-created', { videoId: this.videoId }) |
245 | ||
246 | this.masterWatcher.close() | |
b34ee7fa | 247 | .catch(err => logger.error('Cannot close master watcher of %s.', this.outDirectory, { err, ...this.lTags() })) |
8ebf2a5d C |
248 | }) |
249 | } | |
250 | ||
b34ee7fa | 251 | private watchTSFiles () { |
8ebf2a5d C |
252 | const startStreamDateTime = new Date().getTime() |
253 | ||
b34ee7fa | 254 | this.tsWatcher = watch(this.outDirectory + '/*.ts') |
8ebf2a5d C |
255 | |
256 | const playlistIdMatcher = /^([\d+])-/ | |
257 | ||
75b7117f | 258 | const addHandler = async (segmentPath: string) => { |
4c6757f2 | 259 | logger.debug('Live add handler of %s.', segmentPath, this.lTags()) |
8ebf2a5d C |
260 | |
261 | const playlistId = basename(segmentPath).match(playlistIdMatcher)[0] | |
262 | ||
263 | const segmentsToProcess = this.segmentsToProcessPerPlaylist[playlistId] || [] | |
b34ee7fa | 264 | this.processSegments(segmentsToProcess) |
8ebf2a5d C |
265 | |
266 | this.segmentsToProcessPerPlaylist[playlistId] = [ segmentPath ] | |
267 | ||
268 | if (this.hasClientSocketInBadHealthWithCache(this.sessionId)) { | |
269 | this.emit('bad-socket-health', { videoId: this.videoId }) | |
270 | return | |
271 | } | |
272 | ||
273 | // Duration constraint check | |
274 | if (this.isDurationConstraintValid(startStreamDateTime) !== true) { | |
275 | this.emit('duration-exceeded', { videoId: this.videoId }) | |
276 | return | |
277 | } | |
278 | ||
279 | // Check user quota if the user enabled replay saving | |
280 | if (await this.isQuotaExceeded(segmentPath) === true) { | |
281 | this.emit('quota-exceeded', { videoId: this.videoId }) | |
282 | } | |
283 | } | |
284 | ||
b34ee7fa | 285 | const deleteHandler = (segmentPath: string) => LiveSegmentShaStore.Instance.removeSegmentSha(this.videoUUID, segmentPath) |
8ebf2a5d C |
286 | |
287 | this.tsWatcher.on('add', p => addHandler(p)) | |
288 | this.tsWatcher.on('unlink', p => deleteHandler(p)) | |
289 | } | |
290 | ||
291 | private async isQuotaExceeded (segmentPath: string) { | |
292 | if (this.saveReplay !== true) return false | |
29399256 | 293 | if (this.aborted) return false |
8ebf2a5d C |
294 | |
295 | try { | |
296 | const segmentStat = await stat(segmentPath) | |
297 | ||
298 | LiveQuotaStore.Instance.addQuotaTo(this.user.id, this.videoLive.id, segmentStat.size) | |
299 | ||
300 | const canUpload = await this.isAbleToUploadVideoWithCache(this.user.id) | |
301 | ||
302 | return canUpload !== true | |
303 | } catch (err) { | |
4c6757f2 | 304 | logger.error('Cannot stat %s or check quota of %d.', segmentPath, this.user.id, { err, ...this.lTags() }) |
8ebf2a5d C |
305 | } |
306 | } | |
307 | ||
308 | private createFiles () { | |
309 | for (let i = 0; i < this.allResolutions.length; i++) { | |
310 | const resolution = this.allResolutions[i] | |
311 | ||
312 | const file = new VideoFileModel({ | |
313 | resolution, | |
314 | size: -1, | |
315 | extname: '.ts', | |
316 | infoHash: null, | |
317 | fps: this.fps, | |
318 | videoStreamingPlaylistId: this.streamingPlaylist.id | |
319 | }) | |
320 | ||
321 | VideoFileModel.customUpsert(file, 'streaming-playlist', null) | |
4c6757f2 | 322 | .catch(err => logger.error('Cannot create file for live streaming.', { err, ...this.lTags() })) |
8ebf2a5d C |
323 | } |
324 | } | |
325 | ||
326 | private async prepareDirectories () { | |
4ec52d04 | 327 | await ensureDir(this.outDirectory) |
8ebf2a5d C |
328 | |
329 | if (this.videoLive.saveReplay === true) { | |
4ec52d04 | 330 | await ensureDir(this.replayDirectory) |
8ebf2a5d | 331 | } |
8ebf2a5d C |
332 | } |
333 | ||
334 | private isDurationConstraintValid (streamingStartTime: number) { | |
335 | const maxDuration = CONFIG.LIVE.MAX_DURATION | |
336 | // No limit | |
337 | if (maxDuration < 0) return true | |
338 | ||
339 | const now = new Date().getTime() | |
340 | const max = streamingStartTime + maxDuration | |
341 | ||
342 | return now <= max | |
343 | } | |
344 | ||
b34ee7fa | 345 | private processSegments (segmentPaths: string[]) { |
41fb13c3 | 346 | mapSeries(segmentPaths, async previousSegment => { |
8ebf2a5d C |
347 | // Add sha hash of previous segments, because ffmpeg should have finished generating them |
348 | await LiveSegmentShaStore.Instance.addSegmentSha(this.videoUUID, previousSegment) | |
349 | ||
350 | if (this.saveReplay) { | |
b34ee7fa | 351 | await this.addSegmentToReplay(previousSegment) |
8ebf2a5d | 352 | } |
29399256 C |
353 | }).catch(err => { |
354 | if (this.aborted) return | |
355 | ||
356 | logger.error('Cannot process segments', { err, ...this.lTags() }) | |
357 | }) | |
8ebf2a5d C |
358 | } |
359 | ||
360 | private hasClientSocketInBadHealth (sessionId: string) { | |
361 | const rtmpSession = this.context.sessions.get(sessionId) | |
362 | ||
363 | if (!rtmpSession) { | |
4c6757f2 | 364 | logger.warn('Cannot get session %s to check players socket health.', sessionId, this.lTags()) |
8ebf2a5d C |
365 | return |
366 | } | |
367 | ||
368 | for (const playerSessionId of rtmpSession.players) { | |
369 | const playerSession = this.context.sessions.get(playerSessionId) | |
370 | ||
371 | if (!playerSession) { | |
4c6757f2 | 372 | logger.error('Cannot get player session %s to check socket health.', playerSession, this.lTags()) |
8ebf2a5d C |
373 | continue |
374 | } | |
375 | ||
376 | if (playerSession.socket.writableLength > VIDEO_LIVE.MAX_SOCKET_WAITING_DATA) { | |
377 | return true | |
378 | } | |
379 | } | |
380 | ||
381 | return false | |
382 | } | |
383 | ||
b34ee7fa | 384 | private async addSegmentToReplay (segmentPath: string) { |
8ebf2a5d | 385 | const segmentName = basename(segmentPath) |
4ec52d04 | 386 | const dest = join(this.replayDirectory, buildConcatenatedName(segmentName)) |
8ebf2a5d C |
387 | |
388 | try { | |
389 | const data = await readFile(segmentPath) | |
390 | ||
391 | await appendFile(dest, data) | |
392 | } catch (err) { | |
4c6757f2 | 393 | logger.error('Cannot copy segment %s to replay directory.', segmentPath, { err, ...this.lTags() }) |
8ebf2a5d C |
394 | } |
395 | } | |
396 | } | |
397 | ||
398 | // --------------------------------------------------------------------------- | |
399 | ||
400 | export { | |
401 | MuxingSession | |
402 | } |