import { createServer as createServerTLS, Server as ServerTLS } from 'tls'
import { logger, loggerTagsFactory } from '@server/helpers/logger'
import { CONFIG, registerConfigChangedHandler } from '@server/initializers/config'
-import { VIDEO_LIVE } from '@server/initializers/constants'
+import { VIDEO_LIVE, WEBSERVER } from '@server/initializers/constants'
import { sequelizeTypescript } from '@server/initializers/database'
+import { RunnerJobModel } from '@server/models/runner/runner-job'
import { UserModel } from '@server/models/user/user'
import { VideoModel } from '@server/models/video/video'
import { VideoLiveModel } from '@server/models/video/video-live'
import { LiveQuotaStore } from './live-quota-store'
import { cleanupAndDestroyPermanentLive, getLiveSegmentTime } from './live-utils'
import { MuxingSession } from './shared'
-import { RunnerJobModel } from '@server/models/runner/runner-job'
const NodeRtmpSession = require('node-media-server/src/node_rtmp_session')
const context = require('node-media-server/src/node_core_ctx')
}
const session = this.getContext().sessions.get(sessionId)
+ const inputLocalUrl = session.inputOriginLocalUrl + streamPath
+ const inputPublicUrl = session.inputOriginPublicUrl + streamPath
- this.handleSession(sessionId, session.inputOriginUrl + streamPath, splittedPath[2])
+ this.handleSession({ sessionId, inputPublicUrl, inputLocalUrl, streamKey: splittedPath[2] })
.catch(err => logger.error('Cannot handle sessions.', { err, ...lTags(sessionId) }))
})
events.on('donePublish', sessionId => {
logger.info('Live session ended.', { sessionId, ...lTags(sessionId) })
+
+ // Force session aborting, so we kill ffmpeg even if it still has data to process (slow CPU)
+ setTimeout(() => this.abortSession(sessionId), 2000)
})
registerConfigChangedHandler(() => {
this.rtmpServer = createServer(socket => {
const session = new NodeRtmpSession(config, socket)
- session.inputOriginUrl = 'rtmp://127.0.0.1:' + CONFIG.LIVE.RTMP.PORT
+ session.inputOriginLocalUrl = 'rtmp://127.0.0.1:' + CONFIG.LIVE.RTMP.PORT
+ session.inputOriginPublicUrl = WEBSERVER.RTMP_URL
session.run()
})
this.rtmpsServer = createServerTLS(serverOptions, socket => {
const session = new NodeRtmpSession(config, socket)
- session.inputOriginUrl = 'rtmps://127.0.0.1:' + CONFIG.LIVE.RTMPS.PORT
+ session.inputOriginLocalUrl = 'rtmps://127.0.0.1:' + CONFIG.LIVE.RTMPS.PORT
+ session.inputOriginPublicUrl = WEBSERVER.RTMPS_URL
session.run()
})
return !!this.rtmpServer
}
+ hasSession (sessionId: string) {
+ return this.getContext().sessions.has(sessionId)
+ }
+
stopSessionOf (videoUUID: string, error: LiveVideoError | null) {
const sessionId = this.videoSessions.get(videoUUID)
if (!sessionId) {
}
}
- private async handleSession (sessionId: string, inputUrl: string, streamKey: string) {
+ private async handleSession (options: {
+ sessionId: string
+ inputLocalUrl: string
+ inputPublicUrl: string
+ streamKey: string
+ }) {
+ const { inputLocalUrl, inputPublicUrl, sessionId, streamKey } = options
+
const videoLive = await VideoLiveModel.loadByStreamKey(streamKey)
if (!videoLive) {
logger.warn('Unknown live video with stream key %s.', streamKey, lTags(sessionId))
this.videoSessions.set(video.uuid, sessionId)
const now = Date.now()
- const probe = await ffprobePromise(inputUrl)
+ const probe = await ffprobePromise(inputLocalUrl)
const [ { resolution, ratio }, fps, bitrate, hasAudio ] = await Promise.all([
- getVideoStreamDimensionsInfo(inputUrl, probe),
- getVideoStreamFPS(inputUrl, probe),
- getVideoStreamBitrate(inputUrl, probe),
- hasAudioStream(inputUrl, probe)
+ getVideoStreamDimensionsInfo(inputLocalUrl, probe),
+ getVideoStreamFPS(inputLocalUrl, probe),
+ getVideoStreamBitrate(inputLocalUrl, probe),
+ hasAudioStream(inputLocalUrl, probe)
])
logger.info(
'%s probing took %d ms (bitrate: %d, fps: %d, resolution: %d)',
- inputUrl, Date.now() - now, bitrate, fps, resolution, lTags(sessionId, video.uuid)
+ inputLocalUrl, Date.now() - now, bitrate, fps, resolution, lTags(sessionId, video.uuid)
)
const allResolutions = await Hooks.wrapObject(
sessionId,
videoLive,
- inputUrl,
+ inputLocalUrl,
+ inputPublicUrl,
fps,
bitrate,
ratio,
sessionId: string
videoLive: MVideoLiveVideoWithSetting
- inputUrl: string
+ inputLocalUrl: string
+ inputPublicUrl: string
+
fps: number
bitrate: number
ratio: number
const liveSession = await this.saveStartingSession(videoLive)
const user = await UserModel.loadByLiveId(videoLive.id)
- LiveQuotaStore.Instance.addNewLive(user.id, videoLive.id)
+ LiveQuotaStore.Instance.addNewLive(user.id, sessionId)
const muxingSession = new MuxingSession({
context: this.getContext(),
videoLive,
user,
- ...pick(options, [ 'inputUrl', 'bitrate', 'ratio', 'fps', 'allResolutions', 'hasAudio' ])
+ ...pick(options, [ 'inputLocalUrl', 'inputPublicUrl', 'bitrate', 'ratio', 'fps', 'allResolutions', 'hasAudio' ])
})
muxingSession.on('live-ready', () => this.publishAndFederateLive(videoLive, localLTags))
muxingSession.on('after-cleanup', ({ videoUUID }) => {
this.muxingSessions.delete(sessionId)
- LiveQuotaStore.Instance.removeLive(user.id, videoLive.id)
+ LiveQuotaStore.Instance.removeLive(user.id, sessionId)
muxingSession.destroy()
}
PeerTubeSocket.Instance.sendVideoLiveNewState(video)
+
+ Hooks.runAction('action:live.video.state.updated', { video })
} catch (err) {
logger.error('Cannot save/federate live video %d.', videoId, { err, ...localLTags })
}
}
private onMuxingFFmpegEnd (videoUUID: string, sessionId: string) {
+ // Session already cleaned up
+ if (!this.videoSessions.has(videoUUID)) return
+
this.videoSessions.delete(videoUUID)
this.saveEndingSession(videoUUID, null)
PeerTubeSocket.Instance.sendVideoLiveNewState(fullVideo)
await federateVideoIfNeeded(fullVideo, false)
+
+ Hooks.runAction('action:live.video.state.updated', { video: fullVideo })
} catch (err) {
logger.error('Cannot save/federate new video state of live streaming of video %s.', videoUUID, { err, ...lTags(videoUUID) })
}