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