Refactor live manager

4 files changed, 334 insertions, 225 deletions

diff --git a/server/tests/api/live/index.ts b/server/tests/api/live/index.ts
index c733f564e..e6bcef49f 100644
--- a/server/tests/api/live/index.ts
+++ b/server/tests/api/live/index.ts
@@ -1,4 +1,6 @@
 import './live-constraints'
+import './live-socket-messages'
 import './live-permanent'
 import './live-save-replay'
+import './live-views'
 import './live'
diff --git a/server/tests/api/live/live-socket-messages.ts b/server/tests/api/live/live-socket-messages.ts
new file mode 100644
index 000000000..e00909ade
--- /dev/null
+++ b/server/tests/api/live/live-socket-messages.ts
@@ -0,0 +1,196 @@
+/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
+
+import 'mocha'
+import * as chai from 'chai'
+import { getLiveNotificationSocket } from '@shared/extra-utils/socket/socket-io'
+import { VideoPrivacy, VideoState } from '@shared/models'
+import {
+  cleanupTests,
+  createLive,
+  doubleFollow,
+  flushAndRunMultipleServers,
+  getVideoIdFromUUID,
+  sendRTMPStreamInVideo,
+  ServerInfo,
+  setAccessTokensToServers,
+  setDefaultVideoChannel,
+  stopFfmpeg,
+  updateCustomSubConfig,
+  viewVideo,
+  wait,
+  waitJobs,
+  waitUntilLiveEnded,
+  waitUntilLivePublishedOnAllServers
+} from '../../../../shared/extra-utils'
+
+const expect = chai.expect
+
+describe('Test live', function () {
+  let servers: ServerInfo[] = []
+
+  before(async function () {
+    this.timeout(120000)
+
+    servers = await flushAndRunMultipleServers(2)
+
+    // Get the access tokens
+    await setAccessTokensToServers(servers)
+    await setDefaultVideoChannel(servers)
+
+    await updateCustomSubConfig(servers[0].url, servers[0].accessToken, {
+      live: {
+        enabled: true,
+        allowReplay: true,
+        transcoding: {
+          enabled: false
+        }
+      }
+    })
+
+    // Server 1 and server 2 follow each other
+    await doubleFollow(servers[0], servers[1])
+  })
+
+  describe('Live socket messages', function () {
+
+    async function createLiveWrapper () {
+      const liveAttributes = {
+        name: 'live video',
+        channelId: servers[0].videoChannel.id,
+        privacy: VideoPrivacy.PUBLIC
+      }
+
+      const res = await createLive(servers[0].url, servers[0].accessToken, liveAttributes)
+      return res.body.video.uuid
+    }
+
+    it('Should correctly send a message when the live starts and ends', async function () {
+      this.timeout(60000)
+
+      const localStateChanges: VideoState[] = []
+      const remoteStateChanges: VideoState[] = []
+
+      const liveVideoUUID = await createLiveWrapper()
+      await waitJobs(servers)
+
+      {
+        const videoId = await getVideoIdFromUUID(servers[0].url, liveVideoUUID)
+
+        const localSocket = getLiveNotificationSocket(servers[0].url)
+        localSocket.on('state-change', data => localStateChanges.push(data.state))
+        localSocket.emit('subscribe', { videoId })
+      }
+
+      {
+        const videoId = await getVideoIdFromUUID(servers[1].url, liveVideoUUID)
+
+        const remoteSocket = getLiveNotificationSocket(servers[1].url)
+        remoteSocket.on('state-change', data => remoteStateChanges.push(data.state))
+        remoteSocket.emit('subscribe', { videoId })
+      }
+
+      const command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, liveVideoUUID)
+
+      await waitUntilLivePublishedOnAllServers(servers, liveVideoUUID)
+      await waitJobs(servers)
+
+      for (const stateChanges of [ localStateChanges, remoteStateChanges ]) {
+        expect(stateChanges).to.have.length.at.least(1)
+        expect(stateChanges[stateChanges.length - 1]).to.equal(VideoState.PUBLISHED)
+      }
+
+      await stopFfmpeg(command)
+
+      for (const server of servers) {
+        await waitUntilLiveEnded(server.url, server.accessToken, liveVideoUUID)
+      }
+      await waitJobs(servers)
+
+      for (const stateChanges of [ localStateChanges, remoteStateChanges ]) {
+        expect(stateChanges).to.have.length.at.least(2)
+        expect(stateChanges[stateChanges.length - 1]).to.equal(VideoState.LIVE_ENDED)
+      }
+    })
+
+    it('Should correctly send views change notification', async function () {
+      this.timeout(60000)
+
+      let localLastVideoViews = 0
+      let remoteLastVideoViews = 0
+
+      const liveVideoUUID = await createLiveWrapper()
+      await waitJobs(servers)
+
+      {
+        const videoId = await getVideoIdFromUUID(servers[0].url, liveVideoUUID)
+
+        const localSocket = getLiveNotificationSocket(servers[0].url)
+        localSocket.on('views-change', data => { localLastVideoViews = data.views })
+        localSocket.emit('subscribe', { videoId })
+      }
+
+      {
+        const videoId = await getVideoIdFromUUID(servers[1].url, liveVideoUUID)
+
+        const remoteSocket = getLiveNotificationSocket(servers[1].url)
+        remoteSocket.on('views-change', data => { remoteLastVideoViews = data.views })
+        remoteSocket.emit('subscribe', { videoId })
+      }
+
+      const command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, liveVideoUUID)
+
+      await waitUntilLivePublishedOnAllServers(servers, liveVideoUUID)
+      await waitJobs(servers)
+
+      expect(localLastVideoViews).to.equal(0)
+      expect(remoteLastVideoViews).to.equal(0)
+
+      await viewVideo(servers[0].url, liveVideoUUID)
+      await viewVideo(servers[1].url, liveVideoUUID)
+
+      await waitJobs(servers)
+      await wait(5000)
+      await waitJobs(servers)
+
+      expect(localLastVideoViews).to.equal(2)
+      expect(remoteLastVideoViews).to.equal(2)
+
+      await stopFfmpeg(command)
+    })
+
+    it('Should not receive a notification after unsubscribe', async function () {
+      this.timeout(120000)
+
+      const stateChanges: VideoState[] = []
+
+      const liveVideoUUID = await createLiveWrapper()
+      await waitJobs(servers)
+
+      const videoId = await getVideoIdFromUUID(servers[0].url, liveVideoUUID)
+
+      const socket = getLiveNotificationSocket(servers[0].url)
+      socket.on('state-change', data => stateChanges.push(data.state))
+      socket.emit('subscribe', { videoId })
+
+      const command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, liveVideoUUID)
+
+      await waitUntilLivePublishedOnAllServers(servers, liveVideoUUID)
+      await waitJobs(servers)
+
+      // Notifier waits before sending a notification
+      await wait(10000)
+
+      expect(stateChanges).to.have.lengthOf(1)
+      socket.emit('unsubscribe', { videoId })
+
+      await stopFfmpeg(command)
+      await waitJobs(servers)
+
+      expect(stateChanges).to.have.lengthOf(1)
+    })
+  })
+
+  after(async function () {
+    await cleanupTests(servers)
+  })
+})
diff --git a/server/tests/api/live/live-views.ts b/server/tests/api/live/live-views.ts
new file mode 100644
index 000000000..a44d21ffa
--- /dev/null
+++ b/server/tests/api/live/live-views.ts
@@ -0,0 +1,130 @@
+/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
+
+import 'mocha'
+import * as chai from 'chai'
+import { FfmpegCommand } from 'fluent-ffmpeg'
+import { VideoDetails, VideoPrivacy } from '@shared/models'
+import {
+  cleanupTests,
+  createLive,
+  doubleFollow,
+  flushAndRunMultipleServers,
+  getVideo,
+  sendRTMPStreamInVideo,
+  ServerInfo,
+  setAccessTokensToServers,
+  setDefaultVideoChannel,
+  stopFfmpeg,
+  updateCustomSubConfig,
+  viewVideo,
+  wait,
+  waitJobs,
+  waitUntilLivePublishedOnAllServers
+} from '../../../../shared/extra-utils'
+
+const expect = chai.expect
+
+describe('Test live', function () {
+  let servers: ServerInfo[] = []
+
+  before(async function () {
+    this.timeout(120000)
+
+    servers = await flushAndRunMultipleServers(2)
+
+    // Get the access tokens
+    await setAccessTokensToServers(servers)
+    await setDefaultVideoChannel(servers)
+
+    await updateCustomSubConfig(servers[0].url, servers[0].accessToken, {
+      live: {
+        enabled: true,
+        allowReplay: true,
+        transcoding: {
+          enabled: false
+        }
+      }
+    })
+
+    // Server 1 and server 2 follow each other
+    await doubleFollow(servers[0], servers[1])
+  })
+
+  describe('Live views', function () {
+    let liveVideoId: string
+    let command: FfmpegCommand
+
+    async function countViews (expected: number) {
+      for (const server of servers) {
+        const res = await getVideo(server.url, liveVideoId)
+        const video: VideoDetails = res.body
+
+        expect(video.views).to.equal(expected)
+      }
+    }
+
+    before(async function () {
+      this.timeout(30000)
+
+      const liveAttributes = {
+        name: 'live video',
+        channelId: servers[0].videoChannel.id,
+        privacy: VideoPrivacy.PUBLIC
+      }
+
+      const res = await createLive(servers[0].url, servers[0].accessToken, liveAttributes)
+      liveVideoId = res.body.video.uuid
+
+      command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, liveVideoId)
+      await waitUntilLivePublishedOnAllServers(servers, liveVideoId)
+      await waitJobs(servers)
+    })
+
+    it('Should display no views for a live', async function () {
+      await countViews(0)
+    })
+
+    it('Should view a live twice and display 1 view', async function () {
+      this.timeout(30000)
+
+      await viewVideo(servers[0].url, liveVideoId)
+      await viewVideo(servers[0].url, liveVideoId)
+
+      await wait(7000)
+
+      await waitJobs(servers)
+
+      await countViews(1)
+    })
+
+    it('Should wait and display 0 views', async function () {
+      this.timeout(30000)
+
+      await wait(12000)
+      await waitJobs(servers)
+
+      await countViews(0)
+    })
+
+    it('Should view a live on a remote and on local and display 2 views', async function () {
+      this.timeout(30000)
+
+      await viewVideo(servers[0].url, liveVideoId)
+      await viewVideo(servers[1].url, liveVideoId)
+      await viewVideo(servers[1].url, liveVideoId)
+
+      await wait(7000)
+      await waitJobs(servers)
+
+      await countViews(2)
+    })
+
+    after(async function () {
+      await stopFfmpeg(command)
+    })
+  })
+
+  after(async function () {
+    await cleanupTests(servers)
+  })
+})
diff --git a/server/tests/api/live/live.ts b/server/tests/api/live/live.ts
index 57fb58150..50397924e 100644
--- a/server/tests/api/live/live.ts
+++ b/server/tests/api/live/live.ts
@@ -2,10 +2,8 @@
 
 import 'mocha'
 import * as chai from 'chai'
-import { FfmpegCommand } from 'fluent-ffmpeg'
 import { join } from 'path'
 import { ffprobePromise, getVideoStreamFromFile } from '@server/helpers/ffprobe-utils'
-import { getLiveNotificationSocket } from '@shared/extra-utils/socket/socket-io'
 import { LiveVideo, LiveVideoCreate, Video, VideoDetails, VideoPrivacy, VideoState, VideoStreamingPlaylistType } from '@shared/models'
 import { HttpStatusCode } from '../../../../shared/core-utils/miscs/http-error-codes'
 import {
@@ -22,7 +20,6 @@ import {
   getMyVideosWithFilter,
   getPlaylist,
   getVideo,
-  getVideoIdFromUUID,
   getVideosList,
   getVideosWithFilters,
   killallServers,
@@ -40,11 +37,11 @@ import {
   updateCustomSubConfig,
   updateLive,
   uploadVideoAndGetId,
-  viewVideo,
   wait,
   waitJobs,
   waitUntilLiveEnded,
   waitUntilLivePublished,
+  waitUntilLivePublishedOnAllServers,
   waitUntilLiveSegmentGeneration
 } from '../../../../shared/extra-utils'
 
@@ -53,12 +50,6 @@ const expect = chai.expect
 describe('Test live', function () {
   let servers: ServerInfo[] = []
 
-  async function waitUntilLivePublishedOnAllServers (videoId: string) {
-    for (const server of servers) {
-      await waitUntilLivePublished(server.url, server.accessToken, videoId)
-    }
-  }
-
   before(async function () {
     this.timeout(120000)
 
@@ -247,7 +238,7 @@ describe('Test live', function () {
       liveVideoId = resLive.body.video.uuid
 
       command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, liveVideoId)
-      await waitUntilLivePublishedOnAllServers(liveVideoId)
+      await waitUntilLivePublishedOnAllServers(servers, liveVideoId)
       await waitJobs(servers)
     })
 
@@ -461,7 +452,7 @@ describe('Test live', function () {
       liveVideoId = await createLiveWrapper(false)
 
       const command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, liveVideoId)
-      await waitUntilLivePublishedOnAllServers(liveVideoId)
+      await waitUntilLivePublishedOnAllServers(servers, liveVideoId)
       await waitJobs(servers)
 
       await testVideoResolutions(liveVideoId, [ 720 ])
@@ -477,7 +468,7 @@ describe('Test live', function () {
       liveVideoId = await createLiveWrapper(false)
 
       const command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, liveVideoId)
-      await waitUntilLivePublishedOnAllServers(liveVideoId)
+      await waitUntilLivePublishedOnAllServers(servers, liveVideoId)
       await waitJobs(servers)
 
       await testVideoResolutions(liveVideoId, resolutions)
@@ -494,7 +485,7 @@ describe('Test live', function () {
       liveVideoId = await createLiveWrapper(true)
 
       const command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, liveVideoId, 'video_short2.webm')
-      await waitUntilLivePublishedOnAllServers(liveVideoId)
+      await waitUntilLivePublishedOnAllServers(servers, liveVideoId)
       await waitJobs(servers)
 
       await testVideoResolutions(liveVideoId, resolutions)
@@ -504,7 +495,7 @@ describe('Test live', function () {
 
       await waitJobs(servers)
 
-      await waitUntilLivePublishedOnAllServers(liveVideoId)
+      await waitUntilLivePublishedOnAllServers(servers, liveVideoId)
 
       const bitrateLimits = {
         720: 5000 * 1000, // 60FPS
@@ -559,216 +550,6 @@ describe('Test live', function () {
     })
   })
 
-  describe('Live views', function () {
-    let liveVideoId: string
-    let command: FfmpegCommand
-
-    async function countViews (expected: number) {
-      for (const server of servers) {
-        const res = await getVideo(server.url, liveVideoId)
-        const video: VideoDetails = res.body
-
-        expect(video.views).to.equal(expected)
-      }
-    }
-
-    before(async function () {
-      this.timeout(30000)
-
-      const liveAttributes = {
-        name: 'live video',
-        channelId: servers[0].videoChannel.id,
-        privacy: VideoPrivacy.PUBLIC
-      }
-
-      const res = await createLive(servers[0].url, servers[0].accessToken, liveAttributes)
-      liveVideoId = res.body.video.uuid
-
-      command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, liveVideoId)
-      await waitUntilLivePublishedOnAllServers(liveVideoId)
-      await waitJobs(servers)
-    })
-
-    it('Should display no views for a live', async function () {
-      await countViews(0)
-    })
-
-    it('Should view a live twice and display 1 view', async function () {
-      this.timeout(30000)
-
-      await viewVideo(servers[0].url, liveVideoId)
-      await viewVideo(servers[0].url, liveVideoId)
-
-      await wait(7000)
-
-      await waitJobs(servers)
-
-      await countViews(1)
-    })
-
-    it('Should wait and display 0 views', async function () {
-      this.timeout(30000)
-
-      await wait(7000)
-      await waitJobs(servers)
-
-      await countViews(0)
-    })
-
-    it('Should view a live on a remote and on local and display 2 views', async function () {
-      this.timeout(30000)
-
-      await viewVideo(servers[0].url, liveVideoId)
-      await viewVideo(servers[1].url, liveVideoId)
-      await viewVideo(servers[1].url, liveVideoId)
-
-      await wait(7000)
-      await waitJobs(servers)
-
-      await countViews(2)
-    })
-
-    after(async function () {
-      await stopFfmpeg(command)
-    })
-  })
-
-  describe('Live socket messages', function () {
-
-    async function createLiveWrapper () {
-      const liveAttributes = {
-        name: 'live video',
-        channelId: servers[0].videoChannel.id,
-        privacy: VideoPrivacy.PUBLIC
-      }
-
-      const res = await createLive(servers[0].url, servers[0].accessToken, liveAttributes)
-      return res.body.video.uuid
-    }
-
-    it('Should correctly send a message when the live starts and ends', async function () {
-      this.timeout(60000)
-
-      const localStateChanges: VideoState[] = []
-      const remoteStateChanges: VideoState[] = []
-
-      const liveVideoUUID = await createLiveWrapper()
-      await waitJobs(servers)
-
-      {
-        const videoId = await getVideoIdFromUUID(servers[0].url, liveVideoUUID)
-
-        const localSocket = getLiveNotificationSocket(servers[0].url)
-        localSocket.on('state-change', data => localStateChanges.push(data.state))
-        localSocket.emit('subscribe', { videoId })
-      }
-
-      {
-        const videoId = await getVideoIdFromUUID(servers[1].url, liveVideoUUID)
-
-        const remoteSocket = getLiveNotificationSocket(servers[1].url)
-        remoteSocket.on('state-change', data => remoteStateChanges.push(data.state))
-        remoteSocket.emit('subscribe', { videoId })
-      }
-
-      const command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, liveVideoUUID)
-
-      await waitUntilLivePublishedOnAllServers(liveVideoUUID)
-      await waitJobs(servers)
-
-      for (const stateChanges of [ localStateChanges, remoteStateChanges ]) {
-        expect(stateChanges).to.have.length.at.least(1)
-        expect(stateChanges[stateChanges.length - 1]).to.equal(VideoState.PUBLISHED)
-      }
-
-      await stopFfmpeg(command)
-
-      for (const server of servers) {
-        await waitUntilLiveEnded(server.url, server.accessToken, liveVideoUUID)
-      }
-      await waitJobs(servers)
-
-      for (const stateChanges of [ localStateChanges, remoteStateChanges ]) {
-        expect(stateChanges).to.have.length.at.least(2)
-        expect(stateChanges[stateChanges.length - 1]).to.equal(VideoState.LIVE_ENDED)
-      }
-    })
-
-    it('Should correctly send views change notification', async function () {
-      this.timeout(60000)
-
-      let localLastVideoViews = 0
-      let remoteLastVideoViews = 0
-
-      const liveVideoUUID = await createLiveWrapper()
-      await waitJobs(servers)
-
-      {
-        const videoId = await getVideoIdFromUUID(servers[0].url, liveVideoUUID)
-
-        const localSocket = getLiveNotificationSocket(servers[0].url)
-        localSocket.on('views-change', data => { localLastVideoViews = data.views })
-        localSocket.emit('subscribe', { videoId })
-      }
-
-      {
-        const videoId = await getVideoIdFromUUID(servers[1].url, liveVideoUUID)
-
-        const remoteSocket = getLiveNotificationSocket(servers[1].url)
-        remoteSocket.on('views-change', data => { remoteLastVideoViews = data.views })
-        remoteSocket.emit('subscribe', { videoId })
-      }
-
-      const command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, liveVideoUUID)
-
-      await waitUntilLivePublishedOnAllServers(liveVideoUUID)
-      await waitJobs(servers)
-
-      expect(localLastVideoViews).to.equal(0)
-      expect(remoteLastVideoViews).to.equal(0)
-
-      await viewVideo(servers[0].url, liveVideoUUID)
-      await viewVideo(servers[1].url, liveVideoUUID)
-
-      await waitJobs(servers)
-      await wait(5000)
-      await waitJobs(servers)
-
-      expect(localLastVideoViews).to.equal(2)
-      expect(remoteLastVideoViews).to.equal(2)
-
-      await stopFfmpeg(command)
-    })
-
-    it('Should not receive a notification after unsubscribe', async function () {
-      this.timeout(60000)
-
-      const stateChanges: VideoState[] = []
-
-      const liveVideoUUID = await createLiveWrapper()
-      await waitJobs(servers)
-
-      const videoId = await getVideoIdFromUUID(servers[0].url, liveVideoUUID)
-
-      const socket = getLiveNotificationSocket(servers[0].url)
-      socket.on('state-change', data => stateChanges.push(data.state))
-      socket.emit('subscribe', { videoId })
-
-      const command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, liveVideoUUID)
-
-      await waitUntilLivePublishedOnAllServers(liveVideoUUID)
-      await waitJobs(servers)
-
-      expect(stateChanges).to.have.lengthOf(1)
-      socket.emit('unsubscribe', { videoId })
-
-      await stopFfmpeg(command)
-      await waitJobs(servers)
-
-      expect(stateChanges).to.have.lengthOf(1)
-    })
-  })
-
   describe('After a server restart', function () {
     let liveVideoId: string
     let liveVideoReplayId: string