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