support for spectrum 6.1

packages/app/src/pages/tv/live/[id].jsx

@@ -1,4 +1,5 @@
 import React from "react"
+import Plyr from "plyr"
 import * as antd from "antd"
 import Marquee from "react-fast-marquee"
 import classnames from "classnames"
@@ -9,121 +10,36 @@ import { Icons } from "@components/Icons"
 import LiveChat from "@components/LiveChat"
 import SpectrumModel from "@models/spectrum"
 
-import Plyr from "plyr"
+import * as Decoders from "./decoders"
-import Hls from "hls.js"
-import mpegts from "mpegts.js"
 
 import "plyr/dist/plyr.css"
 import "./index.less"
 
-const DecodersEvents = {
+async function fetchStream(stream_id) {
-	[Hls.Events.FPS_DROP]: (event, data) => {
+	let stream = await SpectrumModel.getStream(stream_id).catch((error) => {
-		console.log("FPS_DROP Detected", data)
-	},
-}
-
-const StreamDecoders = {
-	flv: async (player, source, { onSourceEnd } = {}) => {
-		if (!source) {
-			console.error("Stream source is not defined")
-			return false
-		}
-
-		const decoderInstance = mpegts.createPlayer({
-			type: "flv",
-			isLive: true,
-			enableWorker: true,
-			url: source,
-		})
-
-		if (typeof onSourceEnd === "function") {
-			decoderInstance.on(mpegts.Events.ERROR, onSourceEnd)
-		}
-
-		decoderInstance.attachMediaElement(player)
-
-		decoderInstance.load()
-
-		await decoderInstance.play().catch((error) => {
 		console.error(error)
+		return null
 	})
 
-		return decoderInstance
+	if (!stream) {
-	},
-	hls: (player, source, options = {}) => {
-		if (!player) {
-			console.error("Player is not defined")
 		return false
 	}
 
-		if (!source) {
+	if (Array.isArray(stream)) {
-			console.error("Stream source is not defined")
+		stream = stream[0]
+	}
+
+	if (!stream.sources) {
 		return false
 	}
 
-		const hlsInstance = new Hls({
+	return stream
-			maxLiveSyncPlaybackRate: 1.5,
-			strategy: "bandwidth",
-			autoplay: true,
-			xhrSetup: (xhr) => {
-				if (options.authToken) {
-					xhr.setRequestHeader(
-						"Authorization",
-						`Bearer ${options.authToken}`,
-					)
-				}
-			},
-		})
-
-		if (options.authToken) {
-			source += `?token=${options.authToken}`
-		}
-
-		console.log("Loading media hls >", source, options)
-
-		hlsInstance.attachMedia(player)
-
-		// when media attached, load source
-		hlsInstance.on(Hls.Events.MEDIA_ATTACHED, () => {
-			hlsInstance.loadSource(source)
-		})
-
-		// process quality and tracks levels
-		hlsInstance.on(Hls.Events.MANIFEST_PARSED, (event, data) => {
-			console.log(`${data.levels.length} quality levels found`)
-		})
-
-		// resume to the last position when player resume playback
-		player.addEventListener("play", () => {
-			console.log("Syncing to last position")
-			player.currentTime = hlsInstance.liveSyncPosition
-		})
-
-		// handle errors
-		hlsInstance.on(Hls.Events.ERROR, (event, data) => {
-			console.error(event, data)
-
-			switch (data.details) {
-				case Hls.ErrorDetails.FRAG_LOAD_ERROR: {
-					console.error(`Error loading fragment ${data.frag.url}`)
-					return
-				}
-				default: {
-					return
-				}
-			}
-		})
-
-		// register player decoder events
-		Object.keys(DecodersEvents).forEach((event) => {
-			hlsInstance.on(event, DecodersEvents[event])
-		})
-
-		return hlsInstance
-	},
 }
 
 export default class StreamViewer extends React.Component {
+	static defaultDecoder = "hls"
+	static stateSyncMs = 1 * 60 * 1000 // 1 minute
+
 	state = {
 		isEnded: false,
 		loading: true,
@@ -139,8 +55,8 @@ export default class StreamViewer extends React.Component {
 	videoPlayerRef = React.createRef()
 
 	loadDecoder = async (decoder, ...args) => {
-		if (typeof StreamDecoders[decoder] === "undefined") {
+		if (typeof Decoders[decoder] === "undefined") {
-			console.error("Protocol not supported")
+			console.error("[TV] Protocol not supported")
 			return false
 		}
 
@@ -155,11 +71,9 @@ export default class StreamViewer extends React.Component {
 			this.setState({ decoderInstance: null })
 		}
 
-		console.log(`Switching decoder to: ${decoder}`)
+		console.log(`[TV] Switching decoder to: ${decoder}`)
 
-		const decoderInstance = await StreamDecoders[decoder](...args)
+		const decoderInstance = await Decoders[decoder](...args)
-
-		console.log(decoderInstance)
 
 		await this.setState({
 			decoderInstance: decoderInstance,
@@ -170,30 +84,6 @@ export default class StreamViewer extends React.Component {
 		return decoderInstance
 	}
 
-	loadStream = async (stream_id) => {
-		let stream = await SpectrumModel.getStream(stream_id).catch((error) => {
-			console.error(error)
-			return null
-		})
-
-		if (!stream) {
-			return false
-		}
-
-		if (Array.isArray(stream)) {
-			stream = stream[0]
-		}
-
-		console.log("Stream data >", stream)
-
-		this.setState({
-			stream: stream,
-			spectators: stream.viewers,
-		})
-
-		return stream
-	}
-
 	onSourceEnd = () => {
 		if (typeof this.state.decoderInstance?.destroy === "function") {
 			this.state.decoderInstance.destroy()
@@ -247,53 +137,48 @@ export default class StreamViewer extends React.Component {
 		})
 	}
 
-	componentDidMount = async () => {
+	joinStreamWebsocket = async (stream) => {
-		this.enterPlayerAnimation()
-		this.attachPlayer()
-
-		console.log("custom token> ", this.props.query["token"])
-
-		// load stream
-		const stream = await this.loadStream(this.props.params.id)
-
 		if (!stream) {
-			return this.onSourceEnd()
+			console.error(
-		}
+				`[TV] Cannot connect to stream websocket if no stream provided`,
-
-		// load the flv decoder (by default)
-		if (stream) {
-			if (!stream.sources) {
-				console.error("Stream sources not found")
-				return
-			}
-
-			await this.loadDecoder(
-				"hls",
-				this.videoPlayerRef.current,
-				stream.sources.hls,
-				{
-					onSourceEnd: this.onSourceEnd,
-					authToken: this.props.query["token"],
-				},
 			)
-		}
+			return false
 		}
 
-	componentWillUnmount = () => {
+		const client = await SpectrumModel.createStreamWebsocket(stream._id, {
-		if (typeof this.state.decoderInstance?.unload === "function") {
+			maxConnectRetries: 3,
-			this.state.decoderInstance.unload()
+			refName: "/",
+		})
+
+		this.setState({
+			websocket: client,
+		})
+
+		await client.connect()
+
+		this.streamStateInterval = setInterval(() => {
+			this.syncWithStreamState()
+		}, StreamViewer.stateSyncMs)
+
+		setTimeout(this.syncWithStreamState, 1000)
+
+		return client
 	}
 
-		if (typeof this.state.decoderInstance?.destroy === "function") {
+	syncWithStreamState = async () => {
-			this.state.decoderInstance.destroy()
+		if (!this.state.websocket || !this.state.stream) {
+			return false
 		}
 
-		this.exitPlayerAnimation()
+		const state = await this.state.websocket.requestState()
-		this.toggleCinemaMode(false)
 
-		if (this.streamInfoInterval) {
+		return this.setState({
-			clearInterval(this.streamInfoInterval)
+			spectators: state.viewers,
-		}
+			stream: {
+				...this.state.stream,
+				...(state.profile ?? {}),
+			},
+		})
 	}
 
 	enterPlayerAnimation = () => {
@@ -314,19 +199,7 @@ export default class StreamViewer extends React.Component {
 		}
 	}
 
-	updateQuality = (newQuality) => {
+	setStreamLevel = (level) => {}
-		if (this.state.loadedProtocol !== "hls") {
-			console.error("Unsupported protocol")
-			return false
-		}
-
-		this.state.protocolInstance.levels.forEach((level, levelIndex) => {
-			if (level.height === newQuality) {
-				console.log("Found quality match with " + newQuality)
-				this.state.protocolInstance.currentLevel = levelIndex
-			}
-		})
-	}
 
 	toggleLoading = (to) => {
 		this.setState({ loading: to ?? !this.state.loading })
@@ -343,6 +216,64 @@ export default class StreamViewer extends React.Component {
 		this.setState({ cinemaMode: to })
 	}
 
+	componentDidMount = async () => {
+		this.enterPlayerAnimation()
+		this.attachPlayer()
+
+		// fetch stream data
+		const stream = await fetchStream(this.props.params.id)
+
+		// and error occurred or no stream available/online
+		if (!stream) {
+			return this.onSourceEnd()
+		}
+
+		console.log(`[TV] Stream data >`, stream)
+
+		// set data
+		this.setState({
+			stream: stream,
+			spectators: stream.viewers,
+		})
+
+		// joinStreamWebsocket
+		await this.joinStreamWebsocket(stream)
+
+		// load decoder with provided data
+		await this.loadDecoder(
+			StreamViewer.defaultDecoder,
+			this.videoPlayerRef.current,
+			stream.sources,
+			{
+				onSourceEnd: this.onSourceEnd,
+				authToken: this.props.query["token"],
+			},
+		)
+	}
+
+	componentWillUnmount = () => {
+		if (typeof this.state.decoderInstance?.unload === "function") {
+			this.state.decoderInstance.unload()
+		}
+
+		if (typeof this.state.decoderInstance?.destroy === "function") {
+			this.state.decoderInstance.destroy()
+		}
+
+		if (this.state.websocket) {
+			if (typeof this.state.websocket.destroy === "function") {
+				this.state.websocket.destroy()
+			}
+		}
+
+		this.exitPlayerAnimation()
+		this.toggleCinemaMode(false)
+
+		if (this.streamStateInterval) {
+			clearInterval(this.streamStateInterval)
+		}
+	}
+
 	render() {
 		return (
 			<div

packages/app/src/pages/tv/live/decoders/flv.js

@@ -0,0 +1,29 @@
+import mpegts from "mpegts.js"
+
+export default async (player, source, { onSourceEnd } = {}) => {
+	if (!source) {
+		console.error("Stream source is not defined")
+		return false
+	}
+
+	const decoderInstance = mpegts.createPlayer({
+		type: "flv",
+		isLive: true,
+		enableWorker: true,
+		url: source,
+	})
+
+	if (typeof onSourceEnd === "function") {
+		decoderInstance.on(mpegts.Events.ERROR, onSourceEnd)
+	}
+
+	decoderInstance.attachMediaElement(player)
+
+	decoderInstance.load()
+
+	await decoderInstance.play().catch((error) => {
+		console.error(error)
+	})
+
+	return decoderInstance
+}

packages/app/src/pages/tv/live/decoders/hls.js

@@ -0,0 +1,79 @@
+import Hls from "hls.js"
+
+const Events = {
+	[Hls.Events.FPS_DROP]: (event, data) => {
+		console.warn("[HLS] FPS_DROP Detected", data)
+	},
+	[Hls.Events.ERROR]: (event, data) => {
+		console.error("[HLS] Error", data)
+	},
+}
+
+export default (player, sources = {}, options = {}) => {
+	if (!player) {
+		console.error("[HLS] player is not defined")
+		return false
+	}
+
+	if (!sources.hls) {
+		console.error("[HLS] an hls source is not provided")
+		return false
+	}
+
+	let source = sources.hls
+
+	const hlsInstance = new Hls({
+		maxLiveSyncPlaybackRate: 1.5,
+		strategy: "bandwidth",
+		autoplay: true,
+		xhrSetup: (xhr) => {
+			if (options.authToken) {
+				xhr.setRequestHeader(
+					"Authorization",
+					`Bearer ${options.authToken}`,
+				)
+			}
+		},
+	})
+
+	if (options.authToken) {
+		source += `?token=${options.authToken}`
+	}
+
+	console.log("[HLS] Instance options >", options)
+	console.log(`[HLS] Loading source [${source}]`)
+
+	hlsInstance.attachMedia(player)
+
+	// when media attached, load source
+	hlsInstance.on(Hls.Events.MEDIA_ATTACHED, () => {
+		hlsInstance.loadSource(source)
+	})
+
+	// handle when media ends
+	hlsInstance.on(Hls.Events.BUFFER_EOS, () => {
+		console.log("[HLS] Media ended")
+
+		if (typeof options.onSourceEnd === "function") {
+			options.onSourceEnd()
+		}
+	})
+
+	// process quality and tracks levels
+	hlsInstance.on(Hls.Events.MANIFEST_PARSED, (event, data) => {
+		console.log("[HLS] Manifest parsed >", data)
+	})
+
+	// resume to the last position when player resume playback
+	player.addEventListener("play", () => {
+		console.log("[HLS] Syncing to last position")
+		player.currentTime = hlsInstance.liveSyncPosition
+	})
+
+	// register hls decoder events
+	Object.keys(Events).forEach((event) => {
+		hlsInstance.on(event, Events[event])
+	})
+
+	return hlsInstance
+}

packages/app/src/pages/tv/live/decoders/index.js

@@ -0,0 +1,2 @@
+export { default as hls } from "./hls"
+export { default as flv } from "./flv"

packages/app/src/pages/tv/tabs/livestreamsList/index.jsx

@@ -88,7 +88,7 @@ const LivestreamItem = (props) => {
 
 export default (props) => {
 	const [L_Streams, R_Streams, E_Streams] = app.cores.api.useRequest(
-		SpectrumModel.getLivestreamsList,
+		SpectrumModel.list,
 	)
 
 	useCenteredContainer(false)