]>
Commit | Line | Data |
---|---|---|
1 | ||
2 | import * as Bluebird from 'bluebird' | |
3 | import * as chokidar from 'chokidar' | |
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' | |
8 | import { getLiveMuxingCommand, getLiveTranscodingCommand } from '@server/helpers/ffmpeg-utils' | |
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' | |
14 | import { VideoTranscodingProfilesManager } from '../../transcoding/video-transcoding-profiles' | |
15 | import { isAbleToUploadVideo } from '../../user' | |
16 | import { getHLSDirectory } from '../../video-paths' | |
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 { | |
24 | 'master-playlist-created': ({ videoId: number }) => void | |
25 | ||
26 | 'bad-socket-health': ({ videoId: number }) => void | |
27 | 'duration-exceeded': ({ videoId: number }) => void | |
28 | 'quota-exceeded': ({ videoId: number }) => void | |
29 | ||
30 | 'ffmpeg-end': ({ videoId: number }) => void | |
31 | 'ffmpeg-error': ({ sessionId: string }) => void | |
32 | ||
33 | 'after-cleanup': ({ videoId: number }) => void | |
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 | |
55 | private readonly rtmpUrl: string | |
56 | private readonly fps: number | |
57 | private readonly allResolutions: number[] | |
58 | ||
59 | private readonly videoId: number | |
60 | private readonly videoUUID: string | |
61 | private readonly saveReplay: boolean | |
62 | ||
63 | private readonly lTags: LoggerTagsFn | |
64 | ||
65 | private segmentsToProcessPerPlaylist: { [playlistId: string]: string[] } = {} | |
66 | ||
67 | private tsWatcher: chokidar.FSWatcher | |
68 | private masterWatcher: chokidar.FSWatcher | |
69 | ||
70 | private readonly isAbleToUploadVideoWithCache = memoizee((userId: number) => { | |
71 | return isAbleToUploadVideo(userId, 1000) | |
72 | }, { maxAge: MEMOIZE_TTL.LIVE_ABLE_TO_UPLOAD }) | |
73 | ||
74 | private readonly hasClientSocketInBadHealthWithCache = memoizee((sessionId: string) => { | |
75 | return this.hasClientSocketInBadHealth(sessionId) | |
76 | }, { maxAge: MEMOIZE_TTL.LIVE_CHECK_SOCKET_HEALTH }) | |
77 | ||
78 | constructor (options: { | |
79 | context: any | |
80 | user: MUserId | |
81 | sessionId: string | |
82 | videoLive: MVideoLiveVideo | |
83 | streamingPlaylist: MStreamingPlaylistVideo | |
84 | rtmpUrl: string | |
85 | fps: number | |
86 | allResolutions: number[] | |
87 | }) { | |
88 | super() | |
89 | ||
90 | this.context = options.context | |
91 | this.user = options.user | |
92 | this.sessionId = options.sessionId | |
93 | this.videoLive = options.videoLive | |
94 | this.streamingPlaylist = options.streamingPlaylist | |
95 | this.rtmpUrl = options.rtmpUrl | |
96 | this.fps = options.fps | |
97 | this.allResolutions = options.allResolutions | |
98 | ||
99 | this.videoId = this.videoLive.Video.id | |
100 | this.videoUUID = this.videoLive.Video.uuid | |
101 | ||
102 | this.saveReplay = this.videoLive.saveReplay | |
103 | ||
104 | this.lTags = loggerTagsFactory('live', this.sessionId, this.videoUUID) | |
105 | } | |
106 | ||
107 | async runMuxing () { | |
108 | this.createFiles() | |
109 | ||
110 | const outPath = await this.prepareDirectories() | |
111 | ||
112 | this.ffmpegCommand = CONFIG.LIVE.TRANSCODING.ENABLED | |
113 | ? await getLiveTranscodingCommand({ | |
114 | rtmpUrl: this.rtmpUrl, | |
115 | outPath, | |
116 | resolutions: this.allResolutions, | |
117 | fps: this.fps, | |
118 | availableEncoders: VideoTranscodingProfilesManager.Instance.getAvailableEncoders(), | |
119 | profile: CONFIG.LIVE.TRANSCODING.PROFILE | |
120 | }) | |
121 | : getLiveMuxingCommand(this.rtmpUrl, outPath) | |
122 | ||
123 | logger.info('Running live muxing/transcoding for %s.', this.videoUUID, this.lTags) | |
124 | ||
125 | this.watchTSFiles(outPath) | |
126 | this.watchMasterFile(outPath) | |
127 | ||
128 | this.ffmpegCommand.on('error', (err, stdout, stderr) => { | |
129 | this.onFFmpegError(err, stdout, stderr, outPath) | |
130 | }) | |
131 | ||
132 | this.ffmpegCommand.on('end', () => this.onFFmpegEnded(outPath)) | |
133 | ||
134 | this.ffmpegCommand.run() | |
135 | } | |
136 | ||
137 | abort () { | |
138 | if (!this.ffmpegCommand) return false | |
139 | ||
140 | this.ffmpegCommand.kill('SIGINT') | |
141 | return true | |
142 | } | |
143 | ||
144 | private onFFmpegError (err: any, stdout: string, stderr: string, outPath: string) { | |
145 | this.onFFmpegEnded(outPath) | |
146 | ||
147 | // Don't care that we killed the ffmpeg process | |
148 | if (err?.message?.includes('Exiting normally')) return | |
149 | ||
150 | logger.error('Live transcoding error.', { err, stdout, stderr, ...this.lTags }) | |
151 | ||
152 | this.emit('ffmpeg-error', ({ sessionId: this.sessionId })) | |
153 | } | |
154 | ||
155 | private onFFmpegEnded (outPath: string) { | |
156 | logger.info('RTMP transmuxing for video %s ended. Scheduling cleanup', this.rtmpUrl, this.lTags) | |
157 | ||
158 | setTimeout(() => { | |
159 | // Wait latest segments generation, and close watchers | |
160 | ||
161 | Promise.all([ this.tsWatcher.close(), this.masterWatcher.close() ]) | |
162 | .then(() => { | |
163 | // Process remaining segments hash | |
164 | for (const key of Object.keys(this.segmentsToProcessPerPlaylist)) { | |
165 | this.processSegments(outPath, this.segmentsToProcessPerPlaylist[key]) | |
166 | } | |
167 | }) | |
168 | .catch(err => { | |
169 | logger.error( | |
170 | 'Cannot close watchers of %s or process remaining hash segments.', outPath, | |
171 | { err, ...this.lTags } | |
172 | ) | |
173 | }) | |
174 | ||
175 | this.emit('after-cleanup', { videoId: this.videoId }) | |
176 | }, 1000) | |
177 | } | |
178 | ||
179 | private watchMasterFile (outPath: string) { | |
180 | this.masterWatcher = chokidar.watch(outPath + '/master.m3u8') | |
181 | ||
182 | this.masterWatcher.on('add', async () => { | |
183 | this.emit('master-playlist-created', { videoId: this.videoId }) | |
184 | ||
185 | this.masterWatcher.close() | |
186 | .catch(err => logger.error('Cannot close master watcher of %s.', outPath, { err, ...this.lTags })) | |
187 | }) | |
188 | } | |
189 | ||
190 | private watchTSFiles (outPath: string) { | |
191 | const startStreamDateTime = new Date().getTime() | |
192 | ||
193 | this.tsWatcher = chokidar.watch(outPath + '/*.ts') | |
194 | ||
195 | const playlistIdMatcher = /^([\d+])-/ | |
196 | ||
197 | const addHandler = async segmentPath => { | |
198 | logger.debug('Live add handler of %s.', segmentPath, this.lTags) | |
199 | ||
200 | const playlistId = basename(segmentPath).match(playlistIdMatcher)[0] | |
201 | ||
202 | const segmentsToProcess = this.segmentsToProcessPerPlaylist[playlistId] || [] | |
203 | this.processSegments(outPath, segmentsToProcess) | |
204 | ||
205 | this.segmentsToProcessPerPlaylist[playlistId] = [ segmentPath ] | |
206 | ||
207 | if (this.hasClientSocketInBadHealthWithCache(this.sessionId)) { | |
208 | this.emit('bad-socket-health', { videoId: this.videoId }) | |
209 | return | |
210 | } | |
211 | ||
212 | // Duration constraint check | |
213 | if (this.isDurationConstraintValid(startStreamDateTime) !== true) { | |
214 | this.emit('duration-exceeded', { videoId: this.videoId }) | |
215 | return | |
216 | } | |
217 | ||
218 | // Check user quota if the user enabled replay saving | |
219 | if (await this.isQuotaExceeded(segmentPath) === true) { | |
220 | this.emit('quota-exceeded', { videoId: this.videoId }) | |
221 | } | |
222 | } | |
223 | ||
224 | const deleteHandler = segmentPath => LiveSegmentShaStore.Instance.removeSegmentSha(this.videoUUID, segmentPath) | |
225 | ||
226 | this.tsWatcher.on('add', p => addHandler(p)) | |
227 | this.tsWatcher.on('unlink', p => deleteHandler(p)) | |
228 | } | |
229 | ||
230 | private async isQuotaExceeded (segmentPath: string) { | |
231 | if (this.saveReplay !== true) return false | |
232 | ||
233 | try { | |
234 | const segmentStat = await stat(segmentPath) | |
235 | ||
236 | LiveQuotaStore.Instance.addQuotaTo(this.user.id, this.videoLive.id, segmentStat.size) | |
237 | ||
238 | const canUpload = await this.isAbleToUploadVideoWithCache(this.user.id) | |
239 | ||
240 | return canUpload !== true | |
241 | } catch (err) { | |
242 | logger.error('Cannot stat %s or check quota of %d.', segmentPath, this.user.id, { err, ...this.lTags }) | |
243 | } | |
244 | } | |
245 | ||
246 | private createFiles () { | |
247 | for (let i = 0; i < this.allResolutions.length; i++) { | |
248 | const resolution = this.allResolutions[i] | |
249 | ||
250 | const file = new VideoFileModel({ | |
251 | resolution, | |
252 | size: -1, | |
253 | extname: '.ts', | |
254 | infoHash: null, | |
255 | fps: this.fps, | |
256 | videoStreamingPlaylistId: this.streamingPlaylist.id | |
257 | }) | |
258 | ||
259 | VideoFileModel.customUpsert(file, 'streaming-playlist', null) | |
260 | .catch(err => logger.error('Cannot create file for live streaming.', { err, ...this.lTags })) | |
261 | } | |
262 | } | |
263 | ||
264 | private async prepareDirectories () { | |
265 | const outPath = getHLSDirectory(this.videoLive.Video) | |
266 | await ensureDir(outPath) | |
267 | ||
268 | const replayDirectory = join(outPath, VIDEO_LIVE.REPLAY_DIRECTORY) | |
269 | ||
270 | if (this.videoLive.saveReplay === true) { | |
271 | await ensureDir(replayDirectory) | |
272 | } | |
273 | ||
274 | return outPath | |
275 | } | |
276 | ||
277 | private isDurationConstraintValid (streamingStartTime: number) { | |
278 | const maxDuration = CONFIG.LIVE.MAX_DURATION | |
279 | // No limit | |
280 | if (maxDuration < 0) return true | |
281 | ||
282 | const now = new Date().getTime() | |
283 | const max = streamingStartTime + maxDuration | |
284 | ||
285 | return now <= max | |
286 | } | |
287 | ||
288 | private processSegments (hlsVideoPath: string, segmentPaths: string[]) { | |
289 | Bluebird.mapSeries(segmentPaths, async previousSegment => { | |
290 | // Add sha hash of previous segments, because ffmpeg should have finished generating them | |
291 | await LiveSegmentShaStore.Instance.addSegmentSha(this.videoUUID, previousSegment) | |
292 | ||
293 | if (this.saveReplay) { | |
294 | await this.addSegmentToReplay(hlsVideoPath, previousSegment) | |
295 | } | |
296 | }).catch(err => logger.error('Cannot process segments in %s', hlsVideoPath, { err, ...this.lTags })) | |
297 | } | |
298 | ||
299 | private hasClientSocketInBadHealth (sessionId: string) { | |
300 | const rtmpSession = this.context.sessions.get(sessionId) | |
301 | ||
302 | if (!rtmpSession) { | |
303 | logger.warn('Cannot get session %s to check players socket health.', sessionId, this.lTags) | |
304 | return | |
305 | } | |
306 | ||
307 | for (const playerSessionId of rtmpSession.players) { | |
308 | const playerSession = this.context.sessions.get(playerSessionId) | |
309 | ||
310 | if (!playerSession) { | |
311 | logger.error('Cannot get player session %s to check socket health.', playerSession, this.lTags) | |
312 | continue | |
313 | } | |
314 | ||
315 | if (playerSession.socket.writableLength > VIDEO_LIVE.MAX_SOCKET_WAITING_DATA) { | |
316 | return true | |
317 | } | |
318 | } | |
319 | ||
320 | return false | |
321 | } | |
322 | ||
323 | private async addSegmentToReplay (hlsVideoPath: string, segmentPath: string) { | |
324 | const segmentName = basename(segmentPath) | |
325 | const dest = join(hlsVideoPath, VIDEO_LIVE.REPLAY_DIRECTORY, buildConcatenatedName(segmentName)) | |
326 | ||
327 | try { | |
328 | const data = await readFile(segmentPath) | |
329 | ||
330 | await appendFile(dest, data) | |
331 | } catch (err) { | |
332 | logger.error('Cannot copy segment %s to replay directory.', segmentPath, { err, ...this.lTags }) | |
333 | } | |
334 | } | |
335 | } | |
336 | ||
337 | // --------------------------------------------------------------------------- | |
338 | ||
339 | export { | |
340 | MuxingSession | |
341 | } |