rewrite concat with gstreamer, fix a/v sync problems · stream.place/streamplace@f4be660

+1 -1

Makefile

··· 169 169 -D "gstreamer-full:gst-full-plugins=libgstopusparse.a;libgstcodectimestamper.a;libgstrtp.a;libgstaudioresample.a;libgstlibav.a;libgstmatroska.a;libgstmultifile.a;libgstjpeg.a;libgstaudiotestsrc.a;libgstaudioconvert.a;libgstaudioparsers.a;libgstfdkaac.a;libgstisomp4.a;libgstapp.a;libgstvideoconvertscale.a;libgstvideobox.a;libgstvideorate.a;libgstpng.a;libgstcompositor.a;libgstx264.a;libgstopus.a;libgstvideotestsrc.a;libgstvideoparsersbad.a;libgstaudioparsers.a;libgstmpegtsmux.a;libgstplayback.a;libgsttypefindfunctions.a" \ 170 170 -D "gstreamer-full:gst-full-libraries=gstreamer-controller-1.0,gstreamer-plugins-base-1.0,gstreamer-pbutils-1.0" \ 171 171 -D "gstreamer-full:gst-full-target-type=static_library" \ 172 - -D "gstreamer-full:gst-full-elements=coreelements:concat,filesrc,queue,queue2,multiqueue,typefind,tee,capsfilter,fakesink" \ 172 + -D "gstreamer-full:gst-full-elements=coreelements:concat,filesrc,queue,queue2,multiqueue,typefind,tee,capsfilter,fakesink,identity" \ 173 173 -D "gstreamer-full:bad=enabled" \ 174 174 -D "gstreamer-full:tls=disabled" \ 175 175 -D "gstreamer-full:libav=enabled" \

+83

js/app/components/player/av-sync.tsx

··· 1 + export const QUIET_PROFILE = "audible"; 2 + 3 + export async function quietReceiver( 4 + mediaStream: MediaStream, 5 + playerEvent: (time: string, eventType: string, data: any) => void, 6 + ) { 7 + let audioTime = 0; 8 + let videoTime = 0; 9 + let baseline = 0; 10 + 11 + const diff = (a: number, b: number) => { 12 + if (audioTime === 0 || videoTime === 0) { 13 + return; 14 + } 15 + if (baseline === 0) { 16 + baseline = audioTime - videoTime; 17 + console.log("baseline", baseline); 18 + } 19 + console.log("diff", audioTime - videoTime - baseline); 20 + playerEvent(new Date().toISOString(), "av-sync", { 21 + diff: audioTime - videoTime - baseline, 22 + }); 23 + }; 24 + 25 + const gotVideo = (time: number) => { 26 + videoTime = time; 27 + // diff(audioTime, videoTime); 28 + }; 29 + 30 + const gotAudio = (time: number) => { 31 + audioTime = time; 32 + diff(audioTime, videoTime); 33 + }; 34 + 35 + const Quiet = await import("quietjs-bundle"); 36 + Quiet.addReadyCallback(() => { 37 + const nav = navigator as unknown as any; 38 + // quiet doesn't let us pass in a mediaStream so we need to monkeypatch getusermedia 39 + const getUserMedia = nav.getUserMedia; 40 + nav.getUserMedia = async (constraints, cb) => { 41 + cb(mediaStream); 42 + // we're done, unmonkeypatch 43 + nav.getUserMedia = getUserMedia; 44 + }; 45 + const quiet = Quiet.receiver({ 46 + profile: QUIET_PROFILE, 47 + onReceive: (payload) => { 48 + try { 49 + const str = Quiet.ab2str(payload); 50 + const time = parseInt(str); 51 + gotAudio(time); 52 + } catch (e) { 53 + console.error("quiet receiver error", e); 54 + } 55 + }, 56 + onCreate: () => { 57 + console.log("receiver created"); 58 + }, 59 + onCreateFail: (error) => { 60 + console.error("receiver failed to create", error); 61 + }, 62 + onReceiveFail: (error) => { 63 + console.error("receiver failed to receive", error); 64 + }, 65 + // onReceiverStatsUpdate: (stats) => { 66 + // console.log("receiver stats", stats); 67 + // }, 68 + }); 69 + }); 70 + 71 + const zxing = await import("@zxing/browser"); 72 + const codeReader = new zxing.BrowserQRCodeReader(); 73 + codeReader.decodeFromStream(mediaStream, undefined, (result, err) => { 74 + try { 75 + if (result) { 76 + const time = parseInt(result.getText()); 77 + gotVideo(time); 78 + } 79 + } catch (e) { 80 + console.error("zxing error", e); 81 + } 82 + }); 83 + }

+5 -3

js/app/components/player/controls.tsx

··· 36 36 PROTOCOL_PROGRESSIVE_WEBM, 37 37 PROTOCOL_WEBRTC, 38 38 } from "./props"; 39 - import { selectPlayer, usePlayerActions } from "features/player/playerSlice"; 39 + import { usePlayer, usePlayerActions } from "features/player/playerSlice"; 40 40 import { useAppDispatch, useAppSelector } from "store/hooks"; 41 41 import Loading from "components/loading/loading"; 42 42 ··· 213 213 } 214 214 215 215 function LiveBubble() { 216 - const player = useAppSelector(selectPlayer); 216 + const player = useAppSelector(usePlayer()); 217 217 const dispatch = useAppDispatch(); 218 218 const { startIngest } = usePlayerActions(); 219 219 return ( ··· 226 226 > 227 227 <Button 228 228 backgroundColor="rgba(0,0,0,0.9)" 229 + borderWidth={1} 230 + borderColor="white" 229 231 borderRadius={9999999999} 230 232 padding="$2" 231 233 paddingLeft="$3" ··· 241 243 } 242 244 243 245 function LiveBubbleText() { 244 - const player = useAppSelector(selectPlayer); 246 + const player = useAppSelector(usePlayer()); 245 247 if (!player.ingestStarting) { 246 248 return <H3>START STREAMING</H3>; 247 249 }

+4

js/app/components/player/player.tsx

··· 5 5 import { Text, View } from "tamagui"; 6 6 import Fullscreen from "./fullscreen"; 7 7 import { 8 + IngestMediaSource, 8 9 PlayerEvent, 9 10 PlayerProps, 10 11 PlayerStatus, ··· 125 126 setStatus: setStatus, 126 127 playTime: playTime, 127 128 setPlayTime: setPlayTime, 129 + ingestMediaSource: props.ingestMediaSource ?? IngestMediaSource.USER, 130 + ingestAutoStart: props.ingestAutoStart ?? false, 131 + ...props, 128 132 }; 129 133 return ( 130 134 <View f={1} justifyContent="center" position="relative">

+9

js/app/components/player/props.tsx

··· 1 + export enum IngestMediaSource { 2 + USER = "user", 3 + DISPLAY = "display", 4 + } 5 + 1 6 // common types shared by players and controls and stuff 2 7 export type PlayerProps = { 3 8 name: string; ··· 23 28 playTime: number; 24 29 setPlayTime: (playTime: number) => void; 25 30 ingest?: boolean; 31 + ingestMediaSource?: IngestMediaSource; 32 + ingestStreamKey?: string; 33 + ingestAutoStart?: boolean; 34 + avSyncTest?: boolean; 26 35 }; 27 36 28 37 export type PlayerEvent = {

+9 -9

js/app/components/player/use-webrtc.tsx

··· 150 150 }); 151 151 } 152 152 153 - export function useWebRTCIngest( 154 - endpoint: string, 155 - ): [MediaStream | null, (MediaStream) => void] { 153 + export function useWebRTCIngest({ 154 + endpoint, 155 + streamKey, 156 + }: { 157 + endpoint: string; 158 + streamKey?: string; 159 + }): [MediaStream | null, (mediaStream: MediaStream | null) => void] { 156 160 const [mediaStream, setMediaStream] = useState<MediaStream | null>(null); 157 161 const { ingestConnectionState } = usePlayerActions(); 158 162 const dispatch = useAppDispatch(); 159 - const storedKey = useAppSelector(selectStoredKey); 163 + const storedKey = streamKey ?? useAppSelector(selectStoredKey)?.privateKey; 160 164 useEffect(() => { 161 165 if (storedKey) { 162 166 return; ··· 185 189 } 186 190 }); 187 191 peerConnection.addEventListener("negotiationneeded", (ev) => { 188 - negotiateConnectionWithClientOffer( 189 - peerConnection, 190 - endpoint, 191 - storedKey.privateKey, 192 - ); 192 + negotiateConnectionWithClientOffer(peerConnection, endpoint, storedKey); 193 193 }); 194 194 195 195 return () => {

+56 -20

js/app/components/player/video.tsx

··· 9 9 } from "react"; 10 10 import { View } from "tamagui"; 11 11 import { 12 + IngestMediaSource, 12 13 PlayerProps, 13 14 PlayerStatus, 14 15 PROTOCOL_HLS, ··· 19 20 import { srcToUrl } from "./shared"; 20 21 import useWebRTC, { useWebRTCIngest } from "./use-webrtc"; 21 22 import useAquareumNode from "hooks/useAquareumNode"; 22 - import { selectPlayer } from "features/player/playerSlice"; 23 23 import { useAppDispatch, useAppSelector } from "store/hooks"; 24 24 import { selectStoredKey } from "features/bluesky/blueskySlice"; 25 + import { usePlayer } from "features/player/playerSlice"; 26 + import streamKey from "src/screens/stream-key"; 27 + import { quietReceiver } from "./av-sync"; 25 28 26 29 type VideoProps = PlayerProps & { url: string }; 27 30 ··· 201 204 const [mediaStream] = useWebRTC(props.url); 202 205 203 206 useEffect(() => { 207 + if (!props.avSyncTest) { 208 + return; 209 + } 210 + if (!mediaStream) { 211 + return; 212 + } 213 + quietReceiver(mediaStream, props.playerEvent); 214 + }, [props.avSyncTest, mediaStream]); 215 + 216 + useEffect(() => { 204 217 if (!videoElement) { 205 218 return; 206 219 } ··· 214 227 props: VideoProps & { videoRef: RefObject<HTMLVideoElement> }, 215 228 ) { 216 229 const dispatch = useAppDispatch(); 217 - const player = useAppSelector(selectPlayer); 230 + const player = useAppSelector(usePlayer()); 218 231 const storedKey = useAppSelector(selectStoredKey); 219 232 const [videoElement, setVideoElement] = useState<HTMLVideoElement | null>( 220 233 null, ··· 229 242 const [localMediaStream, setLocalMediaStream] = useState<MediaStream | null>( 230 243 null, 231 244 ); 232 - const [remoteMediaStream, setRemoteMediaStream] = useWebRTCIngest( 233 - `${url}/api/ingest/webrtc`, 234 - ); 245 + const [remoteMediaStream, setRemoteMediaStream] = useWebRTCIngest({ 246 + endpoint: `${url}/api/ingest/webrtc`, 247 + streamKey: props.ingestStreamKey, 248 + }); 235 249 236 250 useEffect(() => { 237 - navigator.mediaDevices 238 - .getUserMedia({ 239 - audio: true, 240 - video: { 241 - width: { min: 200, ideal: 1920, max: 3840 }, 242 - height: { min: 200, ideal: 1080, max: 2160 }, 243 - }, 244 - }) 245 - .then((stream) => { 246 - setLocalMediaStream(stream); 247 - }); 248 - }, []); 251 + if (props.ingestMediaSource === IngestMediaSource.DISPLAY) { 252 + navigator.mediaDevices 253 + .getDisplayMedia({ 254 + audio: true, 255 + video: true, 256 + }) 257 + .then((stream) => { 258 + setLocalMediaStream(stream); 259 + }) 260 + .catch((e) => { 261 + console.error("error getting display media", e); 262 + }); 263 + } else { 264 + navigator.mediaDevices 265 + .getUserMedia({ 266 + audio: true, 267 + video: { 268 + width: { min: 200, ideal: 1920, max: 3840 }, 269 + height: { min: 200, ideal: 1080, max: 2160 }, 270 + }, 271 + }) 272 + .then((stream) => { 273 + setLocalMediaStream(stream); 274 + }) 275 + .catch((e) => { 276 + console.error("error getting user media", e); 277 + }); 278 + } 279 + }, [props.ingestMediaSource]); 249 280 250 281 useEffect(() => { 251 - if (!player.ingestStarting) { 282 + if (!player.ingestStarting && !props.ingestAutoStart) { 252 283 setRemoteMediaStream(null); 253 284 return; 254 285 } 255 286 if (!localMediaStream) { 256 287 return; 257 288 } 258 - if (!storedKey) { 289 + if (!streamKey) { 259 290 return; 260 291 } 261 292 setRemoteMediaStream(localMediaStream); 262 - }, [localMediaStream, player.ingestStarting, storedKey]); 293 + }, [ 294 + localMediaStream, 295 + player.ingestStarting, 296 + streamKey, 297 + props.ingestAutoStart, 298 + ]); 263 299 264 300 useEffect(() => { 265 301 if (!videoElement) {

+5 -2

js/app/features/player/playerSlice.tsx

··· 95 95 }, 96 96 97 97 selectors: { 98 - selectPlayer: (state) => { 99 - const playerId = usePlayerId(); 98 + selectPlayer: (state, playerId: string) => { 100 99 return state[playerId]; 101 100 }, 102 101 }, ··· 119 118 120 119 // Action creators are generated for each case reducer function. 121 120 export const { selectPlayer } = playerSlice.selectors; 121 + export const usePlayer = () => { 122 + const playerId = usePlayerId(); 123 + return (state) => state.player[playerId]; 124 + };

-38

js/app/metro.config.js

··· 38 38 config.resolver.assetExts.push("md"); 39 39 40 40 module.exports = config; 41 - 42 - // REMOVE THIS (just for tamagui internal devs to work in monorepo): 43 - // if (process.env.IS_TAMAGUI_DEV && __dirname.includes('tamagui')) { 44 - // const fs = require('fs') 45 - // const path = require('path') 46 - // const projectRoot = __dirname 47 - // const monorepoRoot = path.resolve(projectRoot, '../..') 48 - // config.watchFolders = [monorepoRoot] 49 - // config.resolver.nodeModulesPaths = [ 50 - // path.resolve(projectRoot, 'node_modules'), 51 - // path.resolve(monorepoRoot, 'node_modules'), 52 - // ] 53 - // // have to manually de-deupe 54 - // try { 55 - // fs.rmSync(path.join(projectRoot, 'node_modules', '@tamagui'), { 56 - // recursive: true, 57 - // force: true, 58 - // }) 59 - // } catch {} 60 - // try { 61 - // fs.rmSync(path.join(projectRoot, 'node_modules', 'tamagui'), { 62 - // recursive: true, 63 - // force: true, 64 - // }) 65 - // } catch {} 66 - // try { 67 - // fs.rmSync(path.join(projectRoot, 'node_modules', 'react'), { 68 - // recursive: true, 69 - // force: true, 70 - // }) 71 - // } catch {} 72 - // try { 73 - // fs.rmSync(path.join(projectRoot, 'node_modules', 'react-dom'), { 74 - // recursive: true, 75 - // force: true, 76 - // }) 77 - // } catch {} 78 - // }

+5

js/app/package.json

··· 43 43 "@tamagui/lucide-icons": "^1.116.12", 44 44 "@tamagui/toast": "^1.116.12", 45 45 "@tanstack/react-query": "^5.59.19", 46 + "@zxing/browser": "^0.1.5", 47 + "@zxing/library": "^0.21.3", 46 48 "abortcontroller-polyfill": "^1.7.6", 47 49 "babel-preset-expo": "~12.0.0", 48 50 "burnt": "^0.12.2", ··· 65 67 "hls.js": "^1.5.17", 66 68 "jose": "^5.9.6", 67 69 "multiformats": "^13.3.1", 70 + "qrcode": "^1.5.4", 71 + "quietjs-bundle": "^0.1.3", 68 72 "react": "18.3.1", 69 73 "react-dom": "18.3.1", 70 74 "react-native": "0.76.2", ··· 99 103 "@tamagui/babel-plugin": "^1.116.12", 100 104 "@tamagui/metro-plugin": "^1.116.12", 101 105 "@types/babel__plugin-transform-runtime": "^7", 106 + "@types/qrcode": "^1", 102 107 "@types/react": "~18.3.12", 103 108 "@types/uuid": "^10.0.0", 104 109 "typescript": "~5.3.3"

+11

js/app/src/router.tsx

··· 39 39 import WebcamScreen from "./screens/webcam"; 40 40 import StreamKeyScreen from "./screens/stream-key"; 41 41 import { hydrate, selectHydrated } from "features/base/baseSlice"; 42 + import AVSyncScreen from "./screens/av-sync"; 42 43 function HomeScreen() { 43 44 return ( 44 45 <View f={1}> ··· 68 69 Webcam: "live/webcam", 69 70 StreamKey: "live/stream-key", 70 71 Login: "login", 72 + AVSync: "sync-test", 71 73 AppReturn: "app-return/:scheme", 72 74 }, 73 75 }, ··· 255 257 options={{ 256 258 drawerLabel: () => null, 257 259 drawerItemStyle: { display: "none" }, 260 + }} 261 + /> 262 + <Drawer.Screen 263 + name="AVSync" 264 + component={AVSyncScreen} 265 + options={{ 266 + drawerLabel: () => null, 267 + drawerItemStyle: { display: "none" }, 268 + headerShown: false, 258 269 }} 259 270 /> 260 271 <Drawer.Screen

+5

js/app/src/screens/av-sync.native.tsx

··· 1 + import { View } from "tamagui"; 2 + 3 + export default function AVSync() { 4 + return <View></View>; 5 + }

+66

js/app/src/screens/av-sync.tsx

··· 1 + import { useEffect, useRef } from "react"; 2 + import { View } from "tamagui"; 3 + import QRCode from "qrcode"; 4 + import { Countdown } from "components"; 5 + import { str2ab } from "quietjs-bundle"; 6 + import { QUIET_PROFILE } from "components/player/av-sync"; 7 + 8 + // screen that displays timestamp as a QR code and encodes timestamp in audio 9 + // so we can measure sync between them 10 + 11 + export default function AVSyncScreen() { 12 + useEffect(() => { 13 + let interval: NodeJS.Timeout | null = null; 14 + async function initQuiet() { 15 + const quiet = await import("quietjs-bundle"); 16 + quiet.addReadyCallback(() => { 17 + const transmitter = quiet.transmitter({ 18 + profile: QUIET_PROFILE, 19 + }); 20 + interval = setInterval(() => { 21 + transmitter.transmit(str2ab(`${Date.now()}`)); 22 + }, 1000); 23 + }); 24 + } 25 + initQuiet(); 26 + return () => { 27 + if (interval) { 28 + clearInterval(interval); 29 + } 30 + }; 31 + }, []); 32 + 33 + const canvasRef = useRef<HTMLCanvasElement>(null); 34 + useEffect(() => { 35 + let stopped = false; 36 + const frame = () => { 37 + if (stopped) { 38 + return; 39 + } 40 + if (canvasRef.current) { 41 + QRCode.toCanvas(canvasRef.current, `${Date.now()}`, function (error) { 42 + if (error) console.error(error); 43 + }); 44 + } 45 + requestAnimationFrame(frame); 46 + }; 47 + frame(); 48 + return () => { 49 + stopped = true; 50 + }; 51 + }, []); 52 + 53 + return ( 54 + <View flex={1} justifyContent="center" alignItems="center"> 55 + <View f={1} justifyContent="center" alignItems="center"> 56 + <Countdown from="now" /> 57 + </View> 58 + <View height={348} f={1} justifyContent="center" alignItems="center"> 59 + <canvas 60 + ref={canvasRef} 61 + style={{ transform: "scale(3)", imageRendering: "pixelated" }} 62 + /> 63 + </View> 64 + </View> 65 + ); 66 + }

+9 -2

js/app/src/screens/stream.tsx

··· 1 1 import { Player } from "components/player/player"; 2 + import { PlayerProps } from "components/player/props"; 3 + import { isWeb } from "tamagui"; 4 + import { queryToProps } from "./util"; 2 5 3 6 export default function StreamScreen({ route }) { 4 7 const { user, protocol, url } = route.params; 8 + let extraProps: Partial<PlayerProps> = {}; 9 + if (isWeb) { 10 + extraProps = queryToProps(new URLSearchParams(window.location.search)); 11 + } 5 12 if (user === "stream") { 6 - return <Player src={url} forceProtocol={protocol} />; 13 + return <Player src={url} forceProtocol={protocol} {...extraProps} />; 7 14 } 8 - return <Player src={user} forceProtocol={protocol} />; 15 + return <Player src={user} forceProtocol={protocol} {...extraProps} />; 9 16 }

+13

js/app/src/screens/util.tsx

··· 1 + import { PlayerProps } from "components/player/props"; 2 + 3 + export const queryToProps = (query: URLSearchParams): Partial<PlayerProps> => { 4 + const entries = { ...Object.fromEntries(query) } as Record<string, any>; 5 + for (const [key, value] of Object.entries(entries)) { 6 + if (value === "true") { 7 + entries[key] = true; 8 + } else if (value === "false") { 9 + entries[key] = false; 10 + } 11 + } 12 + return entries as Partial<PlayerProps>; 13 + };

+5

js/app/src/screens/webcam.native.tsx

··· 1 + import { Player } from "components/player/player"; 2 + 3 + export default function StreamScreen({ route }) { 4 + return <Player ingest={true} src="live" />; 5 + }

+4 -2

js/app/src/screens/webcam.tsx

··· 1 1 import { Player } from "components/player/player"; 2 + import { queryToProps } from "./util"; 2 3 3 - export default function StreamScreen({ route }) { 4 - return <Player ingest={true} src="live" />; 4 + export default function StreamScreen() { 5 + const params = new URLSearchParams(window.location.search); 6 + return <Player ingest={true} src="live" {...queryToProps(params)} />; 5 7 }

+1

js/desktop/package.json

··· 48 48 }, 49 49 "license": "MIT", 50 50 "dependencies": { 51 + "@atproto/crypto": "^0.4.3", 51 52 "electron-squirrel-startup": "^1.0.1", 52 53 "source-map-support": "^0.5.21", 53 54 "update-electron-app": "^3.0.0",

+8

js/desktop/src/env.ts

··· 19 19 updateBaseUrl, 20 20 }; 21 21 } 22 + 23 + // This allows TypeScript to pick up the magic constants that's auto-generated by Forge's Webpack 24 + // plugin that tells the Electron app where to look for the Webpack-bundled app code (depending on 25 + // whether you're running in development or production). 26 + declare const MAIN_WINDOW_WEBPACK_ENTRY: string; 27 + declare const MAIN_WINDOW_PRELOAD_WEBPACK_ENTRY: string; 28 + 29 + export { MAIN_WINDOW_WEBPACK_ENTRY, MAIN_WINDOW_PRELOAD_WEBPACK_ENTRY };

+16 -174

js/desktop/src/index.ts

··· 1 - import { app, BrowserWindow, dialog, globalShortcut, ipcMain } from "electron"; 1 + import { app, dialog, ipcMain } from "electron"; 2 2 import { parseArgs } from "node:util"; 3 - import { resolve } from "path"; 4 3 import "source-map-support/register"; 5 - import { v7 as uuidv7 } from "uuid"; 6 4 import getEnv from "./env"; 7 5 import makeNode from "./node"; 8 6 import initUpdater from "./updater"; 9 7 import { generatePrivateKey, privateKeyToAccount } from "viem/accounts"; 10 - 11 - // This allows TypeScript to pick up the magic constants that's auto-generated by Forge's Webpack 12 - // plugin that tells the Electron app where to look for the Webpack-bundled app code (depending on 13 - // whether you're running in development or production). 14 - declare const MAIN_WINDOW_WEBPACK_ENTRY: string; 15 - declare const MAIN_WINDOW_PRELOAD_WEBPACK_ENTRY: string; 8 + import { makeWindow } from "./window"; 9 + import runTests from "./tests/test-runner"; 10 + import { allTestNames } from "./tests/test-runner"; 16 11 17 12 // Handle creating/removing shortcuts on Windows when installing/uninstalling. 18 13 if (require("electron-squirrel-startup")) { ··· 31 26 type: "string", 32 27 default: "300000", 33 28 }, 29 + "tests-to-run": { 30 + type: "string", 31 + default: allTestNames.join(","), 32 + }, 34 33 }, 35 34 }); 36 35 const env = getEnv(); ··· 50 49 const account = privateKeyToAccount(privateKey); 51 50 const env = { 52 51 AQ_ADMIN_ACCOUNT: account.address.toLowerCase(), 52 + AQ_ALLOWED_STREAMS: account.address.toLowerCase(), 53 53 }; 54 54 try { 55 - if (!args["self-test"]) { 56 - await start(env); 55 + if (args["self-test"]) { 56 + await runTests( 57 + args["tests-to-run"].split(","), 58 + args["self-test-duration"], 59 + privateKey, 60 + ); 57 61 } else { 58 - await runSelfTest(env); 62 + await start(env); 59 63 } 60 64 } catch (e) { 61 65 console.error(e); ··· 72 76 } 73 77 }); 74 78 75 - const makeWindow = async (): Promise<BrowserWindow> => { 76 - const { isDev } = getEnv(); 77 - let logoFile: string; 78 - if (isDev) { 79 - // theoretically cwd is aquareum/js/desktop: 80 - logoFile = resolve(process.cwd(), "assets", "aquareum-logo.png"); 81 - } else { 82 - logoFile = resolve(process.resourcesPath, "aquareum-logo.png"); 83 - } 84 - const window = new BrowserWindow({ 85 - height: 600, 86 - width: 800, 87 - icon: logoFile, 88 - webPreferences: { 89 - preload: MAIN_WINDOW_PRELOAD_WEBPACK_ENTRY, 90 - }, 91 - // titleBarStyle: "hidden", 92 - // titleBarOverlay: true, 93 - }); 94 - 95 - globalShortcut.register("CommandOrControl+Shift+I", () => { 96 - window.webContents.toggleDevTools(); 97 - }); 98 - 99 - globalShortcut.register("CommandOrControl+Shift+R", () => { 100 - window.webContents.reload(); 101 - }); 102 - 103 - window.removeMenu(); 104 - 105 - return window; 106 - }; 107 - 108 79 const start = async (env: { [k: string]: string }): Promise<void> => { 109 80 initUpdater(); 110 81 const { skipNode, nodeFrontend } = getEnv(); ··· 123 94 } 124 95 console.log(`opening ${startPath}`); 125 96 mainWindow.loadURL(startPath); 126 - }; 127 - 128 - const delay = (ms: number) => new Promise((r) => setTimeout(r, ms)); 129 - // how much of our time is spent playing for a success? 130 - const PLAYING_SUCCESS = 0.5; 131 - 132 - const runSelfTest = async (env: { [k: string]: string }): Promise<void> => { 133 - let exitCode = 0; 134 - let done = false; 135 - const { nodeFrontend } = getEnv(); 136 - const { addr, internalAddr, proc } = await makeNode({ 137 - env: { 138 - AQ_TEST_STREAM: "true", 139 - }, 140 - autoQuit: false, 141 - }); 142 - proc.on("exit", () => { 143 - if (!done) { 144 - console.log("node exited early, erroring!"); 145 - app.exit(1); 146 - } 147 - }); 148 - try { 149 - const mainWindow = await makeWindow(); 150 - 151 - const testId = uuidv7(); 152 - const definitions = [ 153 - { 154 - name: "hls", 155 - src: "/hls/stream.m3u8", 156 - }, 157 - { 158 - name: "progressive-mp4", 159 - src: "/stream.mp4", 160 - }, 161 - { 162 - name: "progressive-webm", 163 - src: "/stream.webm", 164 - }, 165 - { 166 - name: "webrtc", 167 - src: "/webrtc", 168 - }, 169 - ]; 170 - const tests = definitions.map((x) => ({ 171 - name: x.name, 172 - playerId: `${testId}-${x.name}`, 173 - src: `${addr}/api/playback/self-test${x.src}`, 174 - showControls: true, 175 - })); 176 - const enc = encodeURIComponent(JSON.stringify(tests)); 177 - 178 - console.log(`http://127.0.0.1:38081/multi/${enc}`); 179 - 180 - let load; 181 - if (nodeFrontend) { 182 - load = `${addr}/multi/${enc}`; 183 - } else { 184 - load = `http://127.0.0.1:38081/multi/${enc}`; 185 - } 186 - console.log(`opening ${load}`); 187 - mainWindow.loadURL(load); 188 - 189 - let foundThumbnail = false; 190 - const interval = setInterval(async () => { 191 - const res = await fetch(`${addr}/api/playback/self-test/stream.jpg`); 192 - if (res.status === 404) { 193 - console.log("no thumbnail found"); 194 - return; 195 - } 196 - if (res.status !== 200) { 197 - console.log( 198 - `unexpected http status ${res.status}, failing thumbnail test`, 199 - ); 200 - clearInterval(interval); 201 - return; 202 - } 203 - const blob = await res.arrayBuffer(); 204 - if (blob.byteLength < 1) { 205 - console.log("thumbnail was empty :("); 206 - return; 207 - } 208 - console.log("found thumbnail!"); 209 - foundThumbnail = true; 210 - clearInterval(interval); 211 - }, 1000); 212 - 213 - await delay(parseInt(args["self-test-duration"])); 214 - clearInterval(interval); 215 - const reports = await Promise.all( 216 - tests.map(async (t) => { 217 - const res = await fetch( 218 - `${internalAddr}/player-report/${t.playerId}`, 219 - ); 220 - const data = (await res.json()) as { [k: string]: number }; 221 - return { ...t, data: data }; 222 - }), 223 - ); 224 - let failed = false; 225 - if (!foundThumbnail) { 226 - console.log("never found a thumbnail, failing test"); 227 - failed = true; 228 - } 229 - const percentages = reports.map((report) => { 230 - let total = 0; 231 - for (const [state, ms] of Object.entries(report.data)) { 232 - total += ms; 233 - } 234 - const pcts: { [k: string]: number } = { playing: 0 }; 235 - for (const [state, ms] of Object.entries(report.data)) { 236 - pcts[state] = ms / total; 237 - } 238 - if (pcts.playing < PLAYING_SUCCESS) { 239 - failed = true; 240 - } 241 - return { ...report, pcts }; 242 - }); 243 - done = true; 244 - console.log(JSON.stringify(percentages, null, 2)); 245 - if (failed) { 246 - console.log("test failed! exiting 1"); 247 - exitCode = 1; 248 - } 249 - } catch (e) { 250 - console.error(`error in self-test: ${e}`); 251 - } finally { 252 - proc.kill("SIGTERM"); 253 - app.exit(exitCode); 254 - } 255 97 }; 256 98 257 99 // This method will be called when Electron has finished

+2 -2

js/desktop/src/node.ts

··· 40 40 autoQuit: boolean; 41 41 }) { 42 42 const exe = await findExe(); 43 - const addr = "127.0.0.1:38082"; 44 - const internalAddr = "127.0.0.1:39092"; 43 + const addr = opts.env.AQ_HTTP_ADDR ?? "127.0.0.1:38082"; 44 + const internalAddr = opts.env.AQ_HTTP_INTERNAL_ADDR ?? "127.0.0.1:39092"; 45 45 const proc = spawn(exe, [], { 46 46 stdio: "inherit", 47 47 env: {

+116

js/desktop/src/tests/playback-test.ts

··· 1 + import { delay, PlayerReport } from "./util"; 2 + import { v7 as uuidv7 } from "uuid"; 3 + import { makeWindow } from "../window"; 4 + import { E2ETest, TestEnv } from "./test-env"; 5 + 6 + const PLAYING_SUCCESS = 0.5; 7 + 8 + export const playbackTest: E2ETest = { 9 + setup: async (testEnv: TestEnv) => { 10 + return { 11 + ...testEnv, 12 + env: { 13 + ...testEnv.env, 14 + AQ_TEST_STREAM: "true", 15 + }, 16 + }; 17 + }, 18 + test: async (testEnv: TestEnv): Promise<string | null> => { 19 + const mainWindow = await makeWindow(); 20 + 21 + const testId = uuidv7(); 22 + const definitions = [ 23 + { 24 + name: "hls", 25 + src: "/hls/stream.m3u8", 26 + }, 27 + { 28 + name: "progressive-mp4", 29 + src: "/stream.mp4", 30 + }, 31 + { 32 + name: "progressive-webm", 33 + src: "/stream.webm", 34 + }, 35 + { 36 + name: "webrtc", 37 + src: "/webrtc", 38 + }, 39 + ]; 40 + const tests = definitions.map((x) => ({ 41 + name: x.name, 42 + playerId: `${testId}-${x.name}`, 43 + src: `${testEnv.addr}/api/playback/self-test${x.src}`, 44 + showControls: true, 45 + })); 46 + const enc = encodeURIComponent(JSON.stringify(tests)); 47 + 48 + const load = `${testEnv.addr}/multi/${enc}`; 49 + 50 + console.log(`opening ${load}`); 51 + mainWindow.loadURL(load); 52 + 53 + let foundThumbnail = false; 54 + const interval = setInterval(async () => { 55 + const res = await fetch( 56 + `${testEnv.addr}/api/playback/self-test/stream.jpg`, 57 + ); 58 + if (res.status === 404) { 59 + console.log("no thumbnail found"); 60 + return; 61 + } 62 + if (res.status !== 200) { 63 + console.log( 64 + `unexpected http status ${res.status}, failing thumbnail test`, 65 + ); 66 + clearInterval(interval); 67 + return; 68 + } 69 + const blob = await res.arrayBuffer(); 70 + if (blob.byteLength < 1) { 71 + console.log("thumbnail was empty :("); 72 + return; 73 + } 74 + console.log("found thumbnail!"); 75 + foundThumbnail = true; 76 + clearInterval(interval); 77 + }, 1000); 78 + 79 + await delay(testEnv.testDuration); 80 + clearInterval(interval); 81 + const reports = await Promise.all( 82 + tests.map(async (t) => { 83 + const res = await fetch( 84 + `${testEnv.internalAddr}/player-report/${t.playerId}`, 85 + ); 86 + const data = (await res.json()) as PlayerReport; 87 + return { ...t, data: data.whatHappened }; 88 + }), 89 + ); 90 + let failed = false; 91 + if (!foundThumbnail) { 92 + console.log("never found a thumbnail, failing test"); 93 + failed = true; 94 + } 95 + const percentages = reports.map((report) => { 96 + let total = 0; 97 + for (const [state, ms] of Object.entries(report.data)) { 98 + total += ms; 99 + } 100 + const pcts: { [k: string]: number } = { playing: 0 }; 101 + for (const [state, ms] of Object.entries(report.data)) { 102 + pcts[state] = ms / total; 103 + } 104 + if (pcts.playing < PLAYING_SUCCESS) { 105 + failed = true; 106 + } 107 + return { ...report, pcts }; 108 + }); 109 + console.log(JSON.stringify(percentages, null, 2)); 110 + await mainWindow.close(); 111 + if (failed) { 112 + return "test failed!"; 113 + } 114 + return null; 115 + }, 116 + };

+110

js/desktop/src/tests/sync-test.ts

··· 1 + import { BrowserWindow, WebFrameMain, webFrameMain, session } from "electron"; 2 + import { MAIN_WINDOW_PRELOAD_WEBPACK_ENTRY } from "../env"; 3 + import { delay, PlayerReport } from "./util"; 4 + import { E2ETest, TestEnv } from "./test-env"; 5 + import { v7 as uuidv7 } from "uuid"; 6 + 7 + const SYNC_TOO_FAR = 20000; 8 + 9 + export const syncTest: E2ETest = { 10 + test: async (testEnv: TestEnv): Promise<string | null> => { 11 + const playerId = `${uuidv7()}-sync-test`; 12 + const window = new BrowserWindow({ 13 + height: 720, 14 + width: 1280, 15 + webPreferences: { 16 + preload: MAIN_WINDOW_PRELOAD_WEBPACK_ENTRY, 17 + }, 18 + title: "aquareum-sync-test", 19 + // titleBarStyle: "hidden", 20 + // titleBarOverlay: true, 21 + }); 22 + let frame: WebFrameMain | undefined; 23 + await new Promise<void>((resolve) => { 24 + window.webContents.on( 25 + "did-frame-navigate", 26 + ( 27 + event, 28 + url, 29 + httpResponseCode, 30 + httpStatusText, 31 + isMainFrame, 32 + frameProcessId, 33 + frameRoutingId, 34 + ) => { 35 + frame = webFrameMain.fromId(frameProcessId, frameRoutingId); 36 + resolve(); 37 + }, 38 + ); 39 + window.loadURL(`${testEnv.addr}/sync-test`); 40 + }); 41 + session.defaultSession.setDisplayMediaRequestHandler( 42 + async (request, callback) => { 43 + callback({ video: frame, audio: frame }); 44 + }, 45 + ); 46 + const streamWindow = new BrowserWindow({ 47 + height: 720, 48 + width: 1280, 49 + x: 0, 50 + y: 0, 51 + webPreferences: { 52 + preload: MAIN_WINDOW_PRELOAD_WEBPACK_ENTRY, 53 + }, 54 + title: "aquareum-sync-stream", 55 + // titleBarStyle: "hidden", 56 + // titleBarOverlay: true, 57 + }); 58 + const streamParams = new URLSearchParams({ 59 + ingestMediaSource: "display", 60 + ingestStreamKey: testEnv.multibaseKey, 61 + ingestAutoStart: "true", 62 + }); 63 + streamWindow.loadURL( 64 + `${testEnv.addr}/live/webcam?${streamParams.toString()}`, 65 + ); 66 + const playbackParams = new URLSearchParams({ 67 + avSyncTest: "true", 68 + playerId: playerId, 69 + }); 70 + const playbackWindow = new BrowserWindow({ 71 + height: 720, 72 + width: 1280, 73 + x: 0, 74 + y: 0, 75 + webPreferences: { 76 + preload: MAIN_WINDOW_PRELOAD_WEBPACK_ENTRY, 77 + }, 78 + title: "aquareum-sync-playback", 79 + // titleBarStyle: "hidden", 80 + // titleBarOverlay: true, 81 + }); 82 + playbackWindow.loadURL( 83 + `${testEnv.addr}/${testEnv.publicAddress}?${playbackParams.toString()}`, 84 + ); 85 + await delay(testEnv.testDuration); 86 + await Promise.all([ 87 + streamWindow.close(), 88 + playbackWindow.close(), 89 + window.close(), 90 + ]); 91 + const res = await fetch( 92 + `${testEnv.internalAddr}/player-report/${playerId}`, 93 + ); 94 + const data = (await res.json()) as PlayerReport; 95 + if (!data.avSync) { 96 + return "av sync not present in output"; 97 + } 98 + console.log(JSON.stringify(data.avSync)); 99 + let problems = []; 100 + for (const f of ["min", "max", "avg"] as const) { 101 + if (Math.abs(data.avSync[f]) > SYNC_TOO_FAR) { 102 + problems.push(`av sync ${f} is ${data.avSync[f]}`); 103 + } 104 + } 105 + if (problems.length > 0) { 106 + return problems.join(", "); 107 + } 108 + return null; 109 + }, 110 + };

+14

js/desktop/src/tests/test-env.ts

··· 1 + export type TestEnv = { 2 + addr: string; 3 + internalAddr: string; 4 + privateKey: string; 5 + publicAddress: string; 6 + multibaseKey: string; 7 + testDuration: number; 8 + env: Record<string, string>; 9 + }; 10 + 11 + export type E2ETest = { 12 + setup?: (testEnv: TestEnv) => Promise<TestEnv>; 13 + test: (testEnv: TestEnv) => Promise<string | null>; 14 + };

+98

js/desktop/src/tests/test-runner.ts

··· 1 + import { bytesToMultibase } from "@atproto/crypto"; 2 + import getEnv from "../env"; 3 + import makeNode from "../node"; 4 + import { playbackTest } from "./playback-test"; 5 + import { syncTest } from "./sync-test"; 6 + import { E2ETest, TestEnv } from "./test-env"; 7 + import { ChildProcess } from "child_process"; 8 + import { privateKeyToAccount } from "viem/accounts"; 9 + import fs from "fs/promises"; 10 + import path from "path"; 11 + import os from "os"; 12 + 13 + const allTests: Record<string, E2ETest> = { 14 + playback: playbackTest, 15 + sync: syncTest, 16 + }; 17 + 18 + export const allTestNames = Object.keys(allTests); 19 + 20 + const randomPort = () => Math.floor(Math.random() * 20000) + 20000; 21 + 22 + export default async function runTests( 23 + tests: string[], 24 + duration: string, 25 + privateKey: `0x${string}`, 26 + ): Promise<void> { 27 + const testFuncs = []; 28 + for (const test of tests) { 29 + if (!allTests[test]) { 30 + throw new Error(`Test ${test} not found`); 31 + } 32 + testFuncs.push(allTests[test]); 33 + } 34 + try { 35 + const results = await Promise.all( 36 + testFuncs.map(async (test) => { 37 + let testProc: ChildProcess | undefined; 38 + try { 39 + const { skipNode } = getEnv(); 40 + const hexKey = privateKey.slice(2); // Remove 0x prefix 41 + const exportedKey = new Uint8Array( 42 + hexKey.match(/.{1,2}/g).map((byte) => parseInt(byte, 16)), 43 + ); 44 + const multibaseKey = bytesToMultibase(exportedKey, "base58btc"); 45 + const account = privateKeyToAccount(privateKey); 46 + const tmpDir = await fs.mkdtemp( 47 + path.join(os.tmpdir(), "aquareum-test-"), 48 + ); 49 + 50 + let testEnv: TestEnv = { 51 + addr: "http://127.0.0.1:38080", 52 + internalAddr: "http://127.0.0.1:39090", 53 + privateKey: privateKey, 54 + publicAddress: account.address.toLowerCase(), 55 + testDuration: parseInt(duration), 56 + multibaseKey, 57 + env: {}, 58 + }; 59 + if (!skipNode) { 60 + testEnv.env = { 61 + AQ_HTTP_ADDR: `127.0.0.1:${randomPort()}`, 62 + AQ_HTTP_INTERNAL_ADDR: `127.0.0.1:${randomPort()}`, 63 + AQ_DATA_DIR: tmpDir, 64 + }; 65 + } 66 + if (test.setup) { 67 + testEnv = await test.setup(testEnv); 68 + } 69 + if (!skipNode) { 70 + const { addr, internalAddr, proc } = await makeNode({ 71 + env: testEnv.env, 72 + autoQuit: false, 73 + }); 74 + testEnv.addr = addr; 75 + testEnv.internalAddr = internalAddr; 76 + testProc = proc; 77 + } 78 + return await test.test(testEnv); 79 + } catch (e) { 80 + console.error("error running test", e.message); 81 + } finally { 82 + if (testProc) { 83 + testProc.kill("SIGTERM"); 84 + } 85 + } 86 + }), 87 + ); 88 + const failures = results.filter((r) => r !== null); 89 + if (failures.length > 0) { 90 + console.error("tests failed", failures.join(", ")); 91 + process.exit(1); 92 + } 93 + process.exit(0); 94 + } catch (e) { 95 + console.error("error running tests", e.message); 96 + process.exit(1); 97 + } 98 + }

+10

js/desktop/src/tests/util.ts

··· 1 + export const delay = (ms: number) => new Promise((r) => setTimeout(r, ms)); 2 + 3 + export type PlayerReport = { 4 + whatHappened: { [k: string]: number }; 5 + avSync?: { 6 + min: number; 7 + max: number; 8 + avg: number; 9 + }; 10 + };

+36

js/desktop/src/window.ts

··· 1 + import { BrowserWindow, globalShortcut } from "electron"; 2 + import getEnv, { MAIN_WINDOW_PRELOAD_WEBPACK_ENTRY } from "./env"; 3 + import { resolve } from "path"; 4 + 5 + export const makeWindow = async (): Promise<BrowserWindow> => { 6 + const { isDev } = getEnv(); 7 + let logoFile: string; 8 + if (isDev) { 9 + // theoretically cwd is aquareum/js/desktop: 10 + logoFile = resolve(process.cwd(), "assets", "aquareum-logo.png"); 11 + } else { 12 + logoFile = resolve(process.resourcesPath, "aquareum-logo.png"); 13 + } 14 + const window = new BrowserWindow({ 15 + height: 600, 16 + width: 800, 17 + icon: logoFile, 18 + webPreferences: { 19 + preload: MAIN_WINDOW_PRELOAD_WEBPACK_ENTRY, 20 + }, 21 + // titleBarStyle: "hidden", 22 + // titleBarOverlay: true, 23 + }); 24 + 25 + globalShortcut.register("CommandOrControl+Shift+I", () => { 26 + window.webContents.toggleDevTools(); 27 + }); 28 + 29 + globalShortcut.register("CommandOrControl+Shift+R", () => { 30 + window.webContents.reload(); 31 + }); 32 + 33 + window.removeMenu(); 34 + 35 + return window; 36 + };

+3 -1

pkg/api/api_internal.go

··· 11 11 "net/http" 12 12 "net/http/pprof" 13 13 "os" 14 + "path/filepath" 14 15 "regexp" 15 16 rtpprof "runtime/pprof" 16 17 "strconv" ··· 107 108 return 108 109 } 109 110 file := <-a.MediaManager.SubscribeSegment(ctx, user) 110 - w.Header().Set("Location", fmt.Sprintf("%s/playback/%s/segment/%s\n", a.CLI.OwnInternalURL(), user, file)) 111 + base := filepath.Base(file) 112 + w.Header().Set("Location", fmt.Sprintf("%s/playback/%s/segment/%s\n", a.CLI.OwnInternalURL(), user, base)) 111 113 w.WriteHeader(301) 112 114 }) 113 115

+8 -13

pkg/api/playback.go

··· 136 136 errors.WriteHTTPBadRequest(w, "user required", nil) 137 137 return 138 138 } 139 - _, err := a.NormalizeUser(ctx, user) 139 + user, err := a.NormalizeUser(ctx, user) 140 140 if err != nil { 141 141 errors.WriteHTTPBadRequest(w, "invalid user", err) 142 142 return ··· 147 147 return 148 148 } 149 149 offer := webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: string(body)} 150 - pr, pw := io.Pipe() 151 - answer, err := media.WebRTCPlayback(ctx, pr, &offer) 150 + answer, err := a.MediaManager.WebRTCPlayback(ctx, user, &offer) 152 151 if err != nil { 153 152 errors.WriteHTTPInternalServerError(w, "error playing back", err) 154 153 return 155 154 } 156 - go func() { 157 - err := a.MediaManager.SegmentToMKV(ctx, user, pw) 158 - if err != nil { 159 - log.Log(ctx, "error writing segment to mkv", err) 160 - } 161 - }() 162 155 w.WriteHeader(201) 163 156 w.Header().Add("Location", r.URL.Path) 164 157 w.Write([]byte(answer.SDP)) ··· 213 206 return 214 207 } 215 208 216 - _, err = atproto.SyncBlueskyRepo(ctx, did, a.Model) 217 - if err != nil { 218 - apierrors.WriteHTTPInternalServerError(w, "could not resolve aquareum key", err) 219 - return 209 + if did != "" { 210 + _, err = atproto.SyncBlueskyRepo(ctx, did, a.Model) 211 + if err != nil { 212 + apierrors.WriteHTTPInternalServerError(w, "could not resolve aquareum key", err) 213 + return 214 + } 220 215 } 221 216 222 217 // user := p.ByName("user")

+9 -1

pkg/cmd/aquareum.go

··· 122 122 cli.StringSliceFlag(fs, &cli.Peers, "peers", "", "other aquareum nodes to replicate to") 123 123 cli.DebugFlag(fs, &cli.Debug, "debug", "", "modified log verbosity for specific functions or files in form func=ToHLS:3,file=gstreamer.go:4") 124 124 fs.BoolVar(&cli.TestStream, "test-stream", false, "run a built-in test stream on boot") 125 + doValidate := fs.Bool("validate", false, "validate media") 125 126 verbosity := fs.String("v", "3", "log verbosity level") 126 127 127 128 fs.Bool("insecure", false, "DEPRECATED, does nothing.") ··· 141 142 if err != nil { 142 143 return err 143 144 } 144 - flag.CommandLine.Parse(nil) 145 + err = flag.CommandLine.Parse(nil) 146 + if err != nil { 147 + return err 148 + } 145 149 vFlag.Value.Set(*verbosity) 146 150 147 151 ctx := context.Background() ··· 157 161 "runtime.Version", runtime.Version()) 158 162 if *version { 159 163 return nil 164 + } 165 + 166 + if *doValidate { 167 + return media.ValidateMedia(ctx) 160 168 } 161 169 162 170 aqhttp.UserAgent = fmt.Sprintf("aquareum/%s", build.Version)

+333

pkg/media/concat.go

··· 1 + package media 2 + 3 + import ( 4 + "context" 5 + "errors" 6 + "fmt" 7 + "io" 8 + "os" 9 + "strings" 10 + "sync" 11 + 12 + "aquareum.tv/aquareum/pkg/log" 13 + "github.com/go-gst/go-gst/gst" 14 + "github.com/go-gst/go-gst/gst/app" 15 + ) 16 + 17 + type ConcatStreamer interface { 18 + SubscribeSegment(ctx context.Context, user string) <-chan string 19 + } 20 + 21 + // This function remains in scope for the duration of a single users' playback 22 + func ConcatStream(ctx context.Context, pipeline *gst.Pipeline, user string, streamer ConcatStreamer) (*gst.Element, <-chan struct{}, error) { 23 + ctx = log.WithLogValues(ctx, "func", "ConcatStream") 24 + ctx, cancel := context.WithCancel(ctx) 25 + 26 + // make 1000000000000 elements! 27 + 28 + // input multiqueue 29 + inputQueue, err := gst.NewElementWithProperties("multiqueue", map[string]any{}) 30 + if err != nil { 31 + return nil, nil, fmt.Errorf("failed to create multiqueue element: %w", err) 32 + } 33 + err = pipeline.Add(inputQueue) 34 + if err != nil { 35 + return nil, nil, fmt.Errorf("failed to add input multiqueue to pipeline: %w", err) 36 + } 37 + for _, tmpl := range inputQueue.GetPadTemplates() { 38 + log.Warn(ctx, "pad template", "name", tmpl.GetName(), "direction", tmpl.Direction()) 39 + } 40 + inputQueuePadVideoSink := inputQueue.GetRequestPad("sink_%u") 41 + if inputQueuePadVideoSink == nil { 42 + return nil, nil, fmt.Errorf("failed to get input queue video sink pad") 43 + } 44 + inputQueuePadAudioSink := inputQueue.GetRequestPad("sink_%u") 45 + if inputQueuePadAudioSink == nil { 46 + return nil, nil, fmt.Errorf("failed to get input queue audio sink pad") 47 + } 48 + inputQueuePadVideoSrc := inputQueue.GetStaticPad("src_0") 49 + if inputQueuePadVideoSrc == nil { 50 + return nil, nil, fmt.Errorf("failed to get input queue video src pad") 51 + } 52 + inputQueuePadAudioSrc := inputQueue.GetStaticPad("src_1") 53 + if inputQueuePadAudioSrc == nil { 54 + return nil, nil, fmt.Errorf("failed to get input queue audio src pad") 55 + } 56 + 57 + // streamsynchronizer 58 + streamsynchronizer, err := gst.NewElementWithProperties("streamsynchronizer", map[string]any{}) 59 + if err != nil { 60 + return nil, nil, fmt.Errorf("failed to create streamsynchronizer element: %w", err) 61 + } 62 + err = pipeline.Add(streamsynchronizer) 63 + if err != nil { 64 + return nil, nil, fmt.Errorf("failed to add streamsynchronizer to pipeline: %w", err) 65 + } 66 + syncPadVideoSink := streamsynchronizer.GetRequestPad("sink_%u") 67 + if syncPadVideoSink == nil { 68 + return nil, nil, fmt.Errorf("failed to get sync video sink pad") 69 + } 70 + syncPadAudioSink := streamsynchronizer.GetRequestPad("sink_%u") 71 + if syncPadAudioSink == nil { 72 + return nil, nil, fmt.Errorf("failed to get sync audio sink pad") 73 + } 74 + syncPadVideoSrc := streamsynchronizer.GetStaticPad("src_0") 75 + if syncPadVideoSrc == nil { 76 + return nil, nil, fmt.Errorf("failed to get sync video src pad") 77 + } 78 + syncPadAudioSrc := streamsynchronizer.GetStaticPad("src_1") 79 + if syncPadAudioSrc == nil { 80 + return nil, nil, fmt.Errorf("failed to get sync audio src pad") 81 + } 82 + 83 + // output multiqueue 84 + outputQueue, err := gst.NewElementWithProperties("multiqueue", map[string]any{}) 85 + if err != nil { 86 + return nil, nil, fmt.Errorf("failed to create multiqueue element: %w", err) 87 + } 88 + err = pipeline.Add(outputQueue) 89 + if err != nil { 90 + return nil, nil, fmt.Errorf("failed to add output multiqueue to pipeline: %w", err) 91 + } 92 + outputQueuePadVideoSink := outputQueue.GetRequestPad("sink_%u") 93 + if outputQueuePadVideoSink == nil { 94 + return nil, nil, fmt.Errorf("failed to get output queue video sink pad") 95 + } 96 + outputQueuePadAudioSink := outputQueue.GetRequestPad("sink_%u") 97 + if outputQueuePadAudioSink == nil { 98 + return nil, nil, fmt.Errorf("failed to get output queue audio sink pad") 99 + } 100 + 101 + // linking 102 + 103 + // input queue to streamsynchronizer 104 + ret := inputQueuePadVideoSrc.Link(syncPadVideoSink) 105 + if ret != gst.PadLinkOK { 106 + return nil, nil, fmt.Errorf("failed to link multiqueue to streamsynchronizer: %v", ret) 107 + } 108 + ret = inputQueuePadAudioSrc.Link(syncPadAudioSink) 109 + if ret != gst.PadLinkOK { 110 + return nil, nil, fmt.Errorf("failed to link multiqueue to streamsynchronizer: %v", ret) 111 + } 112 + 113 + // streamsynchronizer to output queue 114 + ret = syncPadVideoSrc.Link(outputQueuePadVideoSink) 115 + if ret != gst.PadLinkOK { 116 + return nil, nil, fmt.Errorf("failed to link streamsynchronizer to output queue: %v", ret) 117 + } 118 + ret = syncPadAudioSrc.Link(outputQueuePadAudioSink) 119 + if ret != gst.PadLinkOK { 120 + return nil, nil, fmt.Errorf("failed to link streamsynchronizer to output queue: %v", ret) 121 + } 122 + 123 + // ok now we can start looping over input files 124 + 125 + // this goroutine will read all the files from the segment queue and buffer 126 + // them in a pipe so that we don't miss any in between iterations of the output 127 + allFiles := make(chan string, 1024) 128 + go func() { 129 + for { 130 + select { 131 + case <-ctx.Done(): 132 + log.Warn(ctx, "exiting segment reader") 133 + return 134 + case file := <-streamer.SubscribeSegment(ctx, user): 135 + log.Debug(ctx, "got segment", "file", file) 136 + allFiles <- file 137 + if file == "" { 138 + log.Warn(ctx, "no more segments") 139 + return 140 + } 141 + } 142 + } 143 + }() 144 + 145 + // nextFile is the primary loop that pops off a file, creates new demuxer elements for it, 146 + // and pushes into the pipeline 147 + var nextFile func() 148 + nextFile = func() { 149 + pr, pw := io.Pipe() 150 + go func() { 151 + select { 152 + case <-ctx.Done(): 153 + return 154 + case fullpath := <-allFiles: 155 + if fullpath == "" { 156 + log.Warn(ctx, "no more segments") 157 + cancel() 158 + return 159 + } 160 + f, err := os.Open(fullpath) 161 + log.Debug(ctx, "opening segment file", "file", fullpath) 162 + if err != nil { 163 + log.Debug(ctx, "failed to open segment file", "error", err, "file", fullpath) 164 + cancel() 165 + return 166 + } 167 + defer f.Close() 168 + io.Copy(pw, f) 169 + return 170 + } 171 + }() 172 + 173 + demux, err := gst.NewElementWithProperties("qtdemux", map[string]any{}) 174 + if err != nil { 175 + log.Error(ctx, "failed to create demux element", "error", err) 176 + cancel() 177 + return 178 + } 179 + 180 + err = pipeline.Add(demux) 181 + if err != nil { 182 + log.Error(ctx, "failed to add demux to pipeline", "error", err) 183 + cancel() 184 + return 185 + } 186 + 187 + demuxSinkPad := demux.GetStaticPad("sink") 188 + if demuxSinkPad == nil { 189 + log.Error(ctx, "failed to get demux sink pad") 190 + cancel() 191 + return 192 + } 193 + 194 + mu := sync.Mutex{} 195 + count := 0 196 + demux.Connect("pad-added", func(self *gst.Element, pad *gst.Pad) { 197 + mu.Lock() 198 + count += 1 199 + mu.Unlock() 200 + log.Debug(ctx, "demux pad-added", "name", pad.GetName(), "direction", pad.GetDirection()) 201 + var downstreamPad *gst.Pad 202 + if strings.HasPrefix(pad.GetName(), "video_") { 203 + downstreamPad = inputQueuePadVideoSink 204 + } else if strings.HasPrefix(pad.GetName(), "audio_") { 205 + downstreamPad = inputQueuePadAudioSink 206 + } else { 207 + log.Error(ctx, "unknown pad", "name", pad.GetName(), "direction", pad.GetDirection()) 208 + cancel() 209 + return 210 + } 211 + ret := pad.Link(downstreamPad) 212 + if ret != gst.PadLinkOK { 213 + log.Error(ctx, "failed to link demux to downstream pad", "name", pad.GetName(), "direction", pad.GetDirection(), "error", ret) 214 + cancel() 215 + return 216 + } 217 + if pad.GetDirection() == gst.PadDirectionSource { 218 + pad.AddProbe(gst.PadProbeTypeEventBoth, func(pad *gst.Pad, info *gst.PadProbeInfo) gst.PadProbeReturn { 219 + if info.GetEvent().Type() != gst.EventTypeEOS { 220 + return gst.PadProbeOK 221 + } 222 + log.Debug(ctx, "demux EOS", "name", pad.GetName(), "direction", pad.GetDirection()) 223 + pad.Unlink(downstreamPad) 224 + mu.Lock() 225 + defer mu.Unlock() 226 + count -= 1 227 + 228 + if count == 0 { 229 + // don't keep going if our context is done 230 + if ctx.Err() == nil { 231 + nextFile() 232 + } 233 + } 234 + return gst.PadProbeRemove 235 + }) 236 + } 237 + }) 238 + 239 + appsrc, err := gst.NewElementWithProperties("appsrc", map[string]any{ 240 + "is-live": true, 241 + }) 242 + if err != nil { 243 + log.Error(ctx, "failed to get appsrc element from pipeline", "error", err) 244 + cancel() 245 + return 246 + } 247 + err = pipeline.Add(appsrc) 248 + if err != nil { 249 + log.Error(ctx, "failed to add appsrc to pipeline", "error", err) 250 + cancel() 251 + return 252 + } 253 + 254 + demux.SetState(gst.StatePlaying) 255 + appsrc.SetState(gst.StatePlaying) 256 + 257 + src := app.SrcFromElement(appsrc) 258 + 259 + appSrcPad := appsrc.GetStaticPad("src") 260 + if appSrcPad == nil { 261 + log.Error(ctx, "failed to get appsrc pad") 262 + cancel() 263 + return 264 + } 265 + 266 + done := func() { 267 + // appsrc.Unlink(demux) 268 + pads, err := src.GetPads() 269 + if err != nil { 270 + log.Error(ctx, "failed to get pads", "error", err) 271 + cancel() 272 + return 273 + } 274 + for _, pad := range pads { 275 + log.Debug(ctx, "setting pad-idle", "name", pad.GetName(), "direction", pad.GetDirection()) 276 + 277 + pad.AddProbe(gst.PadProbeTypeIdle, func(pad *gst.Pad, info *gst.PadProbeInfo) gst.PadProbeReturn { 278 + log.Debug(ctx, "pad-idle", "name", pad.GetName(), "direction", pad.GetDirection()) 279 + src.EndStream() 280 + return gst.PadProbeRemove 281 + }) 282 + } 283 + } 284 + 285 + src.SetAutomaticEOS(false) 286 + src.SetCallbacks(&app.SourceCallbacks{ 287 + NeedDataFunc: func(self *app.Source, length uint) { 288 + bs := make([]byte, length) 289 + read, err := pr.Read(bs) 290 + if err != nil { 291 + if errors.Is(err, io.EOF) { 292 + if read > 0 { 293 + log.Debug(ctx, "got data on eof???") 294 + cancel() 295 + return 296 + } 297 + log.Debug(ctx, "EOF, ending stream", "length", read) 298 + done() 299 + return 300 + } else { 301 + log.Error(ctx, "failed to read data", "error", err) 302 + cancel() 303 + return 304 + } 305 + } 306 + toPush := bs 307 + if uint(read) < length { 308 + toPush = bs[:read] 309 + } 310 + buffer := gst.NewBufferWithSize(int64(len(toPush))) 311 + buffer.Map(gst.MapWrite).WriteData(toPush) 312 + self.PushBuffer(buffer) 313 + 314 + if uint(read) < length { 315 + log.Debug(ctx, "short write, ending stream", "length", read) 316 + done() 317 + } 318 + }, 319 + }) 320 + 321 + ret := appSrcPad.Link(demuxSinkPad) 322 + if ret != gst.PadLinkOK { 323 + log.Error(ctx, "failed to link appsrc to demux", "error", ret) 324 + cancel() 325 + return 326 + } 327 + } 328 + 329 + // fire it up! 330 + go nextFile() 331 + 332 + return outputQueue, ctx.Done(), nil 333 + }

+221

pkg/media/gstreamer.go

··· 18 18 "github.com/go-gst/go-glib/glib" 19 19 "github.com/go-gst/go-gst/gst" 20 20 "github.com/go-gst/go-gst/gst/app" 21 + "github.com/google/uuid" 21 22 "github.com/skip2/go-qrcode" 22 23 "golang.org/x/sync/errgroup" 23 24 ) ··· 742 743 743 744 return nil 744 745 } 746 + 747 + func (mm *MediaManager) MP4Playback(ctx context.Context, user string, w io.Writer) error { 748 + uu, err := uuid.NewV7() 749 + if err != nil { 750 + return err 751 + } 752 + ctx = log.WithLogValues(ctx, "playbackID", uu.String()) 753 + ctx, cancel := context.WithCancel(ctx) 754 + 755 + ctx = log.WithLogValues(ctx, "mediafunc", "MP4Playback") 756 + 757 + pipelineSlice := []string{ 758 + "mp4mux name=muxer fragment-mode=first-moov-then-finalise fragment-duration=1000 streamable=true ! appsink name=mp4sink", 759 + "h264parse name=videoparse ! muxer.", 760 + "opusparse name=audioparse ! muxer.", 761 + } 762 + 763 + pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n")) 764 + if err != nil { 765 + return fmt.Errorf("failed to create GStreamer pipeline: %w", err) 766 + } 767 + 768 + ok := pipeline.GetPipelineBus().AddWatch(func(msg *gst.Message) bool { 769 + switch msg.Type() { 770 + case gst.MessageEOS: // When end-of-stream is received flush the pipeling and stop the main loop 771 + log.Log(ctx, "got gst.MessageEOS, exiting") 772 + cancel() 773 + case gst.MessageError: // Error messages are always fatal 774 + err := msg.ParseError() 775 + log.Error(ctx, "gstreamer error", "error", err.Error()) 776 + if debug := err.DebugString(); debug != "" { 777 + log.Log(ctx, "gstreamer debug", "message", debug) 778 + } 779 + cancel() 780 + default: 781 + log.Debug(ctx, msg.String()) 782 + } 783 + return true 784 + }) 785 + if !ok { 786 + return fmt.Errorf("failed to add watch to pipeline bus") 787 + } 788 + 789 + outputQueue, done, err := ConcatStream(ctx, pipeline, user, mm) 790 + if err != nil { 791 + return fmt.Errorf("failed to get output queue: %w", err) 792 + } 793 + go func() { 794 + select { 795 + case <-ctx.Done(): 796 + return 797 + case <-done: 798 + cancel() 799 + } 800 + }() 801 + 802 + videoParse, err := pipeline.GetElementByName("videoparse") 803 + if err != nil { 804 + return fmt.Errorf("failed to get video sink element from pipeline: %w", err) 805 + } 806 + err = outputQueue.Link(videoParse) 807 + if err != nil { 808 + return fmt.Errorf("failed to link output queue to video parse: %w", err) 809 + } 810 + 811 + audioParse, err := pipeline.GetElementByName("audioparse") 812 + if err != nil { 813 + return fmt.Errorf("failed to get audio parse element from pipeline: %w", err) 814 + } 815 + err = outputQueue.Link(audioParse) 816 + if err != nil { 817 + return fmt.Errorf("failed to link output queue to audio parse: %w", err) 818 + } 819 + 820 + go func() { 821 + <-ctx.Done() 822 + pipeline.BlockSetState(gst.StateNull) 823 + }() 824 + 825 + go func() { 826 + ticker := time.NewTicker(time.Second * 1) 827 + for { 828 + select { 829 + case <-ctx.Done(): 830 + return 831 + case <-ticker.C: 832 + state := pipeline.GetCurrentState() 833 + log.Debug(ctx, "pipeline state", "state", state) 834 + } 835 + } 836 + }() 837 + 838 + mp4sinkele, err := pipeline.GetElementByName("mp4sink") 839 + if err != nil { 840 + return fmt.Errorf("failed to get video sink element from pipeline: %w", err) 841 + } 842 + mp4sink := app.SinkFromElement(mp4sinkele) 843 + mp4sink.SetCallbacks(&app.SinkCallbacks{ 844 + NewSampleFunc: WriterNewSample(ctx, w), 845 + EOSFunc: func(sink *app.Sink) { 846 + log.Warn(ctx, "mp4sink EOSFunc") 847 + cancel() 848 + }, 849 + }) 850 + 851 + pipeline.SetState(gst.StatePlaying) 852 + 853 + <-ctx.Done() 854 + return nil 855 + } 856 + 857 + func (mm *MediaManager) MKVPlayback(ctx context.Context, user string, w io.Writer) error { 858 + uu, err := uuid.NewV7() 859 + if err != nil { 860 + return err 861 + } 862 + ctx = log.WithLogValues(ctx, "playbackID", uu.String()) 863 + ctx, cancel := context.WithCancel(ctx) 864 + 865 + ctx = log.WithLogValues(ctx, "mediafunc", "MKVPlayback") 866 + 867 + pipelineSlice := []string{ 868 + "matroskamux name=muxer streamable=true ! appsink name=mkvsink", 869 + "h264parse name=videoparse ! muxer.", 870 + "opusparse name=audioparse ! muxer.", 871 + } 872 + 873 + pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n")) 874 + if err != nil { 875 + return fmt.Errorf("failed to create GStreamer pipeline: %w", err) 876 + } 877 + 878 + ok := pipeline.GetPipelineBus().AddWatch(func(msg *gst.Message) bool { 879 + switch msg.Type() { 880 + case gst.MessageEOS: // When end-of-stream is received flush the pipeling and stop the main loop 881 + log.Log(ctx, "got gst.MessageEOS, exiting") 882 + cancel() 883 + case gst.MessageError: // Error messages are always fatal 884 + err := msg.ParseError() 885 + log.Error(ctx, "gstreamer error", "error", err.Error()) 886 + if debug := err.DebugString(); debug != "" { 887 + log.Log(ctx, "gstreamer debug", "message", debug) 888 + } 889 + cancel() 890 + default: 891 + log.Debug(ctx, msg.String()) 892 + } 893 + return true 894 + }) 895 + if !ok { 896 + return fmt.Errorf("failed to add watch to pipeline bus") 897 + } 898 + 899 + outputQueue, done, err := ConcatStream(ctx, pipeline, user, mm) 900 + if err != nil { 901 + return fmt.Errorf("failed to get output queue: %w", err) 902 + } 903 + go func() { 904 + select { 905 + case <-ctx.Done(): 906 + return 907 + case <-done: 908 + cancel() 909 + } 910 + }() 911 + 912 + videoParse, err := pipeline.GetElementByName("videoparse") 913 + if err != nil { 914 + return fmt.Errorf("failed to get video sink element from pipeline: %w", err) 915 + } 916 + err = outputQueue.Link(videoParse) 917 + if err != nil { 918 + return fmt.Errorf("failed to link output queue to video parse: %w", err) 919 + } 920 + 921 + audioParse, err := pipeline.GetElementByName("audioparse") 922 + if err != nil { 923 + return fmt.Errorf("failed to get audio parse element from pipeline: %w", err) 924 + } 925 + err = outputQueue.Link(audioParse) 926 + if err != nil { 927 + return fmt.Errorf("failed to link output queue to audio parse: %w", err) 928 + } 929 + 930 + go func() { 931 + <-ctx.Done() 932 + pipeline.BlockSetState(gst.StateNull) 933 + }() 934 + 935 + go func() { 936 + ticker := time.NewTicker(time.Second * 1) 937 + for { 938 + select { 939 + case <-ctx.Done(): 940 + return 941 + case <-ticker.C: 942 + state := pipeline.GetCurrentState() 943 + log.Debug(ctx, "pipeline state", "state", state) 944 + } 945 + } 946 + }() 947 + 948 + mkvsinkele, err := pipeline.GetElementByName("mkvsink") 949 + if err != nil { 950 + return fmt.Errorf("failed to get video sink element from pipeline: %w", err) 951 + } 952 + mkvsink := app.SinkFromElement(mkvsinkele) 953 + mkvsink.SetCallbacks(&app.SinkCallbacks{ 954 + NewSampleFunc: WriterNewSample(ctx, w), 955 + EOSFunc: func(sink *app.Sink) { 956 + log.Warn(ctx, "mp4sink EOSFunc") 957 + cancel() 958 + }, 959 + }) 960 + 961 + pipeline.SetState(gst.StatePlaying) 962 + 963 + <-ctx.Done() 964 + return nil 965 + }

+5 -5

pkg/media/media.go

··· 8 8 "errors" 9 9 "fmt" 10 10 "io" 11 - "path/filepath" 12 11 "sync" 13 12 14 13 "aquareum.tv/aquareum/pkg/aqtime" ··· 107 106 } 108 107 109 108 // subscribe to the latest segments from a given user for livestreaming purposes 110 - func (mm *MediaManager) SubscribeSegment(ctx context.Context, user string) chan string { 109 + func (mm *MediaManager) SubscribeSegment(ctx context.Context, user string) <-chan string { 111 110 mm.mp4subsmut.Lock() 112 111 defer mm.mp4subsmut.Unlock() 113 112 _, ok := mm.mp4subs[user] ··· 124 123 mm.mp4subsmut.Lock() 125 124 defer mm.mp4subsmut.Unlock() 126 125 for _, sub := range mm.mp4subs[user] { 127 - sub <- file 126 + go func() { 127 + sub <- file 128 + }() 128 129 } 129 130 mm.mp4subs[user] = []chan string{} 130 131 } ··· 365 366 go mm.replicator.NewSegment(ctx, buf) 366 367 r = bytes.NewReader(buf) 367 368 io.Copy(fd, r) 368 - base := filepath.Base(fd.Name()) 369 - go mm.PublishSegment(ctx, pub.String(), base) 369 + go mm.PublishSegment(ctx, pub.String(), fd.Name()) 370 370 seg := &model.Segment{ 371 371 ID: *mani.Label, 372 372 User: pub.String(),

+282

pkg/media/media_validator.go

··· 1 + package media 2 + 3 + import ( 4 + "context" 5 + "fmt" 6 + "strings" 7 + "time" 8 + 9 + "aquareum.tv/aquareum/pkg/log" 10 + "github.com/go-gst/go-gst/gst" 11 + "github.com/go-gst/go-gst/gst/app" 12 + "github.com/pion/webrtc/v4/pkg/media" 13 + ) 14 + 15 + type MediaValidator struct { 16 + idx int 17 + } 18 + 19 + // var files []string = []string{ 20 + // // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-00-411Z.mp4", 21 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-01-212Z.mp4", 22 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-01-830Z.mp4", 23 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-02-492Z.mp4", 24 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-03-163Z.mp4", 25 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-03-430Z.mp4", 26 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-04-209Z.mp4", 27 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-04-604Z.mp4", 28 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-05-308Z.mp4", 29 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-05-970Z.mp4", 30 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-06-406Z.mp4", 31 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-07-271Z.mp4", 32 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-07-868Z.mp4", 33 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-08-572Z.mp4", 34 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-09-286Z.mp4", 35 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-09-404Z.mp4", 36 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-10-289Z.mp4", 37 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-11-431Z.mp4", 38 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-12-390Z.mp4", 39 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-13-585Z.mp4", 40 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-14-588Z.mp4", 41 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-15-409Z.mp4", 42 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-17-372Z.mp4", 43 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-18-407Z.mp4", 44 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-19-025Z.mp4", 45 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-19-591Z.mp4", 46 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-20-369Z.mp4", 47 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-20-967Z.mp4", 48 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-21-393Z.mp4", 49 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-21-970Z.mp4", 50 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-22-812Z.mp4", 51 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-24-391Z.mp4", 52 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-24-988Z.mp4", 53 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-25-606Z.mp4", 54 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-26-310Z.mp4", 55 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-27-333Z.mp4", 56 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-27-452Z.mp4", 57 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-28-305Z.mp4", 58 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-29-052Z.mp4", 59 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-29-607Z.mp4", 60 + // "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/22/18/2025-01-15T22-18-30-407Z.mp4", 61 + // } 62 + 63 + var files []string = []string{ 64 + "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/23/29/2025-01-15T23-29-00-459Z.mp4", // evil 65 + "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/23/29/2025-01-15T23-29-03-424Z.mp4", // good 66 + "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/23/29/2025-01-15T23-29-04-661Z.mp4", // good 67 + "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/23/29/2025-01-15T23-29-07-121Z.mp4", // good 68 + "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/23/29/2025-01-15T23-29-11-285Z.mp4", // good 69 + "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/23/29/2025-01-15T23-29-12-938Z.mp4", // evil 70 + "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/23/29/2025-01-15T23-29-17-343Z.mp4", // evil 71 + "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/23/29/2025-01-15T23-29-19-158Z.mp4", // good 72 + "/home/iameli/.aquareum/segments/0x3371a9b874d9815c8d18e7d4662cda099a4737b2/2025/01/15/23/29/2025-01-15T23-29-22-261Z.mp4", // good 73 + // "/home/iameli/Desktop/out/2025-01-15T23-29-00-459Z.mp4.mkv.mp4", 74 + // "/home/iameli/Desktop/out/2025-01-15T23-29-03-424Z.mp4.mkv.mp4", 75 + // "/home/iameli/Desktop/out/2025-01-15T23-29-04-661Z.mp4.mkv.mp4", 76 + // "/home/iameli/Desktop/out/2025-01-15T23-29-07-121Z.mp4.mkv.mp4", 77 + // "/home/iameli/Desktop/out/2025-01-15T23-29-11-285Z.mp4.mkv.mp4", 78 + // "/home/iameli/Desktop/out/2025-01-15T23-29-12-938Z.mp4.mkv.mp4", 79 + // "/home/iameli/Desktop/out/2025-01-15T23-29-17-343Z.mp4.mkv.mp4", 80 + // "/home/iameli/Desktop/out/2025-01-15T23-29-19-158Z.mp4.mkv.mp4", 81 + // "/home/iameli/Desktop/out/2025-01-15T23-29-22-261Z.mp4.mkv.mp4", 82 + } 83 + 84 + func (mv *MediaValidator) SubscribeSegment(ctx context.Context, user string) <-chan string { 85 + ch := make(chan string, 1024) 86 + go func() { 87 + if mv.idx >= len(files) { 88 + ch <- "" 89 + return 90 + } 91 + ch <- files[mv.idx] 92 + mv.idx += 1 93 + }() 94 + return ch 95 + } 96 + 97 + func ValidateMedia(ctx context.Context) error { 98 + mv := &MediaValidator{} 99 + 100 + ctx, cancel := context.WithCancel(ctx) 101 + 102 + ctx = log.WithLogValues(ctx, "mediafunc", "ValidateMedia") 103 + 104 + log.Debug(ctx, "starting pipeline") 105 + 106 + pipelineSlice := []string{ 107 + "h264timestamper name=videoparse ! h264parse ! capsfilter caps=video/x-h264,stream-format=byte-stream ! appsink name=videoappsink", 108 + "opusparse name=audioparse ! appsink name=audioappsink", 109 + } 110 + 111 + pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n")) 112 + if err != nil { 113 + return fmt.Errorf("failed to create GStreamer pipeline: %w", err) 114 + } 115 + 116 + ok := pipeline.GetPipelineBus().AddWatch(func(msg *gst.Message) bool { 117 + switch msg.Type() { 118 + case gst.MessageEOS: // When end-of-stream is received flush the pipeling and stop the main loop 119 + log.Log(ctx, "got gst.MessageEOS, exiting") 120 + cancel() 121 + case gst.MessageError: // Error messages are always fatal 122 + err := msg.ParseError() 123 + log.Error(ctx, "gstreamer error", "error", err.Error()) 124 + if debug := err.DebugString(); debug != "" { 125 + log.Log(ctx, "gstreamer debug", "message", debug) 126 + } 127 + cancel() 128 + default: 129 + log.Debug(ctx, msg.String()) 130 + } 131 + return true 132 + }) 133 + 134 + if !ok { 135 + return fmt.Errorf("failed to add watch to pipeline bus") 136 + } 137 + 138 + outputQueue, done, err := ConcatStream(ctx, pipeline, "user", mv) 139 + if err != nil { 140 + return fmt.Errorf("failed to get output queue: %w", err) 141 + } 142 + go func() { 143 + select { 144 + case <-ctx.Done(): 145 + return 146 + case <-done: 147 + cancel() 148 + } 149 + }() 150 + // queuePadVideo := outputQueue.GetRequestPad("src_%u") 151 + // if queuePadVideo == nil { 152 + // return fmt.Errorf("failed to get queue video pad") 153 + // } 154 + // queuePadAudio := outputQueue.GetRequestPad("src_%u") 155 + // if queuePadAudio == nil { 156 + // return fmt.Errorf("failed to get queue audio pad") 157 + // } 158 + 159 + videoParse, err := pipeline.GetElementByName("videoparse") 160 + if err != nil { 161 + return fmt.Errorf("failed to get video sink element from pipeline: %w", err) 162 + } 163 + err = outputQueue.Link(videoParse) 164 + if err != nil { 165 + return fmt.Errorf("failed to link output queue to video parse: %w", err) 166 + } 167 + 168 + audioParse, err := pipeline.GetElementByName("audioparse") 169 + if err != nil { 170 + return fmt.Errorf("failed to get audio parse element from pipeline: %w", err) 171 + } 172 + err = outputQueue.Link(audioParse) 173 + if err != nil { 174 + return fmt.Errorf("failed to link output queue to audio parse: %w", err) 175 + } 176 + 177 + go func() { 178 + <-ctx.Done() 179 + pipeline.BlockSetState(gst.StateNull) 180 + }() 181 + 182 + go func() { 183 + ticker := time.NewTicker(time.Second * 1) 184 + for { 185 + select { 186 + case <-ctx.Done(): 187 + return 188 + case <-ticker.C: 189 + state := pipeline.GetCurrentState() 190 + log.Debug(ctx, "pipeline state", "state", state) 191 + } 192 + } 193 + }() 194 + 195 + videoappsinkele, err := pipeline.GetElementByName("videoappsink") 196 + if err != nil { 197 + return fmt.Errorf("failed to get video sink element from pipeline: %w", err) 198 + } 199 + 200 + audioappsinkele, err := pipeline.GetElementByName("audioappsink") 201 + if err != nil { 202 + return fmt.Errorf("failed to get audio sink element from pipeline: %w", err) 203 + } 204 + 205 + videoappsink := app.SinkFromElement(videoappsinkele) 206 + videoappsink.SetCallbacks(&app.SinkCallbacks{ 207 + NewSampleFunc: func(sink *app.Sink) gst.FlowReturn { 208 + sample := sink.PullSample() 209 + if sample == nil { 210 + return gst.FlowEOS 211 + } 212 + 213 + buffer := sample.GetBuffer() 214 + if buffer == nil { 215 + return gst.FlowError 216 + } 217 + 218 + samples := buffer.Map(gst.MapRead).Bytes() 219 + defer buffer.Unmap() 220 + clockTime := buffer.Duration() 221 + dur := clockTime.AsDuration() 222 + 223 + mediaSample := media.Sample{Data: samples} 224 + if dur != nil { 225 + mediaSample.Duration = *dur 226 + } else { 227 + log.Log(ctx, "no video duration", "samples", len(samples), "segment duration", sample.GetSegment().GetDuration()) 228 + // cancel() 229 + return gst.FlowOK 230 + } 231 + 232 + return gst.FlowOK 233 + }, 234 + EOSFunc: func(sink *app.Sink) { 235 + log.Warn(ctx, "videoappsink EOSFunc") 236 + cancel() 237 + }, 238 + }) 239 + 240 + audioappsink := app.SinkFromElement(audioappsinkele) 241 + audioappsink.SetCallbacks(&app.SinkCallbacks{ 242 + NewSampleFunc: func(sink *app.Sink) gst.FlowReturn { 243 + sample := sink.PullSample() 244 + if sample == nil { 245 + return gst.FlowEOS 246 + } 247 + 248 + buffer := sample.GetBuffer() 249 + if buffer == nil { 250 + return gst.FlowError 251 + } 252 + 253 + samples := buffer.Map(gst.MapRead).Bytes() 254 + defer buffer.Unmap() 255 + 256 + clockTime := buffer.Duration() 257 + dur := clockTime.AsDuration() 258 + mediaSample := media.Sample{Data: samples} 259 + if dur != nil { 260 + mediaSample.Duration = *dur 261 + } else { 262 + log.Log(ctx, "no audio duration", "samples", len(samples)) 263 + // cancel() 264 + return gst.FlowOK 265 + } 266 + 267 + return gst.FlowOK 268 + }, 269 + EOSFunc: func(sink *app.Sink) { 270 + log.Warn(ctx, "audioappsink EOSFunc") 271 + cancel() 272 + }, 273 + }) 274 + 275 + // Start the pipeline 276 + pipeline.SetState(gst.StatePlaying) 277 + log.Warn(ctx, "playing pipeline") 278 + 279 + <-ctx.Done() 280 + log.Warn(ctx, "!!!!!!!!!!!!!!!!!!!!!!! ctx done") 281 + return nil 282 + }

+90 -28

pkg/media/webrtc.go

··· 3 3 import ( 4 4 "context" 5 5 "fmt" 6 - "io" 7 6 "strings" 8 7 "time" 9 8 ··· 18 17 "github.com/pion/webrtc/v4/pkg/media" 19 18 ) 20 19 20 + // we have a bug that prevents us from correctly probing video durations 21 + // a lot of the time. so when we don't have them we use the last duration 22 + // that we had, and when we don't have that we use a default duration 23 + var DEFAULT_DURATION = time.Duration(32 * time.Millisecond) 24 + 21 25 // This function remains in scope for the duration of a single users' playback 22 - func WebRTCPlayback(ctx context.Context, input io.Reader, offer *webrtc.SessionDescription) (*webrtc.SessionDescription, error) { 26 + func (mm *MediaManager) WebRTCPlayback(ctx context.Context, user string, offer *webrtc.SessionDescription) (*webrtc.SessionDescription, error) { 23 27 uu, err := uuid.NewV7() 24 28 if err != nil { 25 29 return nil, err ··· 30 34 ctx = log.WithLogValues(ctx, "mediafunc", "WebRTCPlayback") 31 35 32 36 pipelineSlice := []string{ 33 - "appsrc name=appsrc ! matroskademux name=demux", 34 - "multiqueue name=queue", 35 - "demux.video_0 ! queue.sink_0", 36 - "demux.audio_0 ! queue.sink_1", 37 - "multiqueue name=outqueue", 38 - "queue.src_0 ! h264parse name=videoparse ! video/x-h264,stream-format=byte-stream ! appsink name=videoappsink", 39 - "queue.src_1 ! opusparse ! appsink name=audioappsink", 37 + "h264parse name=videoparse ! video/x-h264,stream-format=byte-stream ! appsink name=videoappsink", 38 + "opusparse name=audioparse ! appsink name=audioappsink", 40 39 } 41 40 42 41 pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n")) ··· 44 43 return nil, fmt.Errorf("failed to create GStreamer pipeline: %w", err) 45 44 } 46 45 47 - pipeline.GetPipelineBus().AddWatch(func(msg *gst.Message) bool { 46 + ok := pipeline.GetPipelineBus().AddWatch(func(msg *gst.Message) bool { 48 47 switch msg.Type() { 49 - 50 48 case gst.MessageEOS: // When end-of-stream is received flush the pipeling and stop the main loop 51 49 log.Log(ctx, "got gst.MessageEOS, exiting") 52 50 cancel() ··· 62 60 } 63 61 return true 64 62 }) 63 + if !ok { 64 + return nil, fmt.Errorf("failed to add watch to pipeline bus") 65 + } 65 66 66 - appsrc, err := pipeline.GetElementByName("appsrc") 67 + outputQueue, done, err := ConcatStream(ctx, pipeline, user, mm) 67 68 if err != nil { 68 - return nil, fmt.Errorf("failed to get appsrc element from pipeline: %w", err) 69 + return nil, fmt.Errorf("failed to get output queue: %w", err) 69 70 } 70 - 71 - src := app.SrcFromElement(appsrc) 72 - src.SetCallbacks(&app.SourceCallbacks{ 73 - NeedDataFunc: ReaderNeedData(ctx, input), 74 - }) 75 - 76 71 go func() { 77 - <-ctx.Done() 78 - pipeline.BlockSetState(gst.StateNull) 72 + select { 73 + case <-ctx.Done(): 74 + return 75 + case <-done: 76 + cancel() 77 + } 79 78 }() 79 + // queuePadVideo := outputQueue.GetRequestPad("src_%u") 80 + // if queuePadVideo == nil { 81 + // return nil, fmt.Errorf("failed to get queue video pad") 82 + // } 83 + // queuePadAudio := outputQueue.GetRequestPad("src_%u") 84 + // if queuePadAudio == nil { 85 + // return nil, fmt.Errorf("failed to get queue audio pad") 86 + // } 87 + 88 + videoParse, err := pipeline.GetElementByName("videoparse") 89 + if err != nil { 90 + return nil, fmt.Errorf("failed to get video sink element from pipeline: %w", err) 91 + } 92 + err = outputQueue.Link(videoParse) 93 + if err != nil { 94 + return nil, fmt.Errorf("failed to link output queue to video parse: %w", err) 95 + } 96 + 97 + audioParse, err := pipeline.GetElementByName("audioparse") 98 + if err != nil { 99 + return nil, fmt.Errorf("failed to get audio parse element from pipeline: %w", err) 100 + } 101 + err = outputQueue.Link(audioParse) 102 + if err != nil { 103 + return nil, fmt.Errorf("failed to link output queue to audio parse: %w", err) 104 + } 80 105 81 106 videoappsinkele, err := pipeline.GetElementByName("videoappsink") 82 107 if err != nil { ··· 146 171 // Setup complete! Now we boot up streaming in the background while returning the SDP offer to the user. 147 172 148 173 go func() { 174 + <-ctx.Done() 175 + pipeline.BlockSetState(gst.StateNull) 176 + }() 177 + 178 + go func() { 179 + ticker := time.NewTicker(time.Second * 1) 180 + for { 181 + select { 182 + case <-ctx.Done(): 183 + return 184 + case <-ticker.C: 185 + state := pipeline.GetCurrentState() 186 + log.Debug(ctx, "pipeline state", "state", state) 187 + } 188 + } 189 + }() 190 + 191 + var lastVideoDuration = &DEFAULT_DURATION 192 + 193 + go func() { 149 194 150 195 videoappsink := app.SinkFromElement(videoappsinkele) 151 196 videoappsink.SetCallbacks(&app.SinkCallbacks{ ··· 162 207 163 208 samples := buffer.Map(gst.MapRead).Bytes() 164 209 defer buffer.Unmap() 210 + clockTime := buffer.Duration() 211 + dur := clockTime.AsDuration() 212 + mediaSample := media.Sample{Data: samples} 213 + if dur != nil { 214 + mediaSample.Duration = *dur 215 + lastVideoDuration = dur 216 + } else if lastVideoDuration != nil { 217 + mediaSample.Duration = *lastVideoDuration 218 + } else { 219 + log.Log(ctx, "no video duration", "samples", len(samples)) 220 + // cancel() 221 + return gst.FlowOK 222 + } 165 223 166 - if err := videoTrack.WriteSample(media.Sample{Data: samples, Duration: *buffer.Duration().AsDuration()}); err != nil { 224 + if err := videoTrack.WriteSample(mediaSample); err != nil { 167 225 log.Log(ctx, "failed to write video sample", "error", err) 168 226 cancel() 169 227 } ··· 171 229 return gst.FlowOK 172 230 }, 173 231 EOSFunc: func(sink *app.Sink) { 232 + log.Warn(ctx, "videoappsink EOSFunc") 174 233 cancel() 175 234 }, 176 235 }) ··· 197 256 if dur != nil { 198 257 mediaSample.Duration = *dur 199 258 } else { 200 - log.Log(ctx, "no duration", "samples", len(samples)) 259 + log.Log(ctx, "no audio duration", "samples", len(samples)) 201 260 // cancel() 202 261 return gst.FlowOK 203 262 } ··· 209 268 return gst.FlowOK 210 269 }, 211 270 EOSFunc: func(sink *app.Sink) { 271 + log.Warn(ctx, "audioappsink EOSFunc") 212 272 cancel() 213 273 }, 214 274 }) ··· 258 318 }) 259 319 260 320 <-ctx.Done() 321 + log.Warn(ctx, "!!!!!!!!!!!!!!!!!!!!!!! ctx done") 261 322 }() 262 323 select { 263 324 case <-gatherComplete: ··· 342 403 343 404 pipelineSlice := []string{ 344 405 "multiqueue name=queue", 345 - "appsrc format=time is-live=true do-timestamp=true name=videosrc ! capsfilter caps=application/x-rtp ! rtph264depay ! capsfilter caps=video/x-h264,stream-format=byte-stream,alignment=nal ! h264parse ! h264timestamper ! queue.sink_0", 406 + "appsrc format=time is-live=true do-timestamp=true name=videosrc ! capsfilter caps=application/x-rtp ! rtph264depay ! capsfilter caps=video/x-h264,stream-format=byte-stream,alignment=nal ! h264parse ! h264timestamper ! identity ! queue.sink_0", 346 407 "appsrc format=time is-live=true do-timestamp=true name=audiosrc ! capsfilter caps=application/x-rtp,media=audio,encoding-name=OPUS,payload=111 ! rtpopusdepay ! queue.sink_1", 347 408 } 348 409 ··· 355 416 switch msg.Type() { 356 417 357 418 case gst.MessageEOS: // When end-of-stream is received flush the pipeling and stop the main loop 358 - log.Log(ctx, "got gst.MessageEOS, exiting") 419 + log.Debug(ctx, "got gst.MessageEOS, exiting") 359 420 cancel() 360 421 case gst.MessageError: // Error messages are always fatal 361 422 err := msg.ParseError() 362 423 log.Error(ctx, "gstreamer error", "error", err.Error()) 363 424 if debug := err.DebugString(); debug != "" { 364 - log.Log(ctx, "gstreamer debug", "message", debug) 425 + log.Debug(ctx, "gstreamer debug", "message", debug) 365 426 } 366 427 cancel() 367 428 default: 368 - log.Log(ctx, msg.String()) 429 + log.Debug(ctx, msg.String()) 369 430 } 370 431 return true 371 432 }) ··· 461 522 return 462 523 case <-ticker.C: 463 524 state := pipeline.GetCurrentState() 464 - log.Log(ctx, "pipeline state", "state", state) 525 + log.Debug(ctx, "pipeline state", "state", state) 465 526 } 466 527 } 467 528 }() ··· 580 641 }) 581 642 582 643 <-ctx.Done() 644 + log.Warn(ctx, "!!!!!!!!! context done, exiting") 583 645 }() 584 646 select { 585 647 case <-gatherComplete:

+1 -1

pkg/model/model.go

··· 25 25 26 26 CreatePlayerEvent(event PlayerEventAPI) error 27 27 ListPlayerEvents(playerId string) ([]PlayerEvent, error) 28 - PlayerReport(playerId string) (map[string]float64, error) 28 + PlayerReport(playerId string) (map[string]any, error) 29 29 ClearPlayerEvents() error 30 30 31 31 CreateSegment(segment *Segment) error

+51 -3

pkg/model/player_event.go

··· 4 4 "database/sql" 5 5 "encoding/json" 6 6 "fmt" 7 + "math" 7 8 "time" 8 9 9 10 "github.com/google/uuid" ··· 65 66 return events, nil 66 67 } 67 68 68 - func (m *DBModel) PlayerReport(playerId string) (map[string]float64, error) { 69 + func (m *DBModel) PlayerReport(playerId string) (map[string]any, error) { 69 70 events, err := m.ListPlayerEvents(playerId) 70 71 if err != nil { 71 72 return nil, err 72 73 } 73 - report := map[string]float64{} 74 + whatHappenedReport := map[string]float64{} 74 75 for _, e := range events { 75 76 if e.EventType != "aq-played" { 76 77 continue ··· 95 96 for state, time := range whatHappened { 96 97 ms, ok := time.(float64) 97 98 if ok { 98 - report[state] = report[state] + ms 99 + whatHappenedReport[state] = whatHappenedReport[state] + ms 99 100 } 100 101 } 101 102 } 103 + 104 + avSyncs := []float64{} 105 + for _, e := range events { 106 + if e.EventType != "av-sync" { 107 + continue 108 + } 109 + bs, err := e.Meta.MarshalJSON() 110 + if err != nil { 111 + return nil, err 112 + } 113 + meta := map[string]any{} 114 + err = json.Unmarshal(bs, &meta) 115 + if err != nil { 116 + return nil, err 117 + } 118 + diff, ok := meta["diff"].(float64) 119 + if !ok { 120 + continue 121 + } 122 + avSyncs = append(avSyncs, diff) 123 + } 124 + 125 + report := map[string]any{ 126 + "whatHappened": whatHappenedReport, 127 + } 128 + 129 + if len(avSyncs) > 0 { 130 + min := math.Inf(1) 131 + max := math.Inf(-1) 132 + sum := 0.0 133 + for _, sync := range avSyncs { 134 + if sync < min { 135 + min = sync 136 + } 137 + if sync > max { 138 + max = sync 139 + } 140 + sum += sync 141 + } 142 + avg := sum / float64(len(avSyncs)) 143 + report["avSync"] = map[string]float64{ 144 + "min": min, 145 + "max": max, 146 + "avg": avg, 147 + } 148 + } 149 + 102 150 return report, nil 103 151 } 104 152

+169 -2

yarn.lock

··· 214 214 languageName: node 215 215 linkType: hard 216 216 217 + "@atproto/crypto@npm:^0.4.3": 218 + version: 0.4.3 219 + resolution: "@atproto/crypto@npm:0.4.3" 220 + dependencies: 221 + "@noble/curves": "npm:^1.7.0" 222 + "@noble/hashes": "npm:^1.6.1" 223 + uint8arrays: "npm:3.0.0" 224 + checksum: 10/387f1ca50621569ad72f0f682fc64b18f6531103ce6dd2aaf422a6aafd4708bc6a393ee9651c000aed984d6d6864b13448aeed492eaa65d586c03b612bbdbfeb 225 + languageName: node 226 + linkType: hard 227 + 217 228 "@atproto/did@npm:0.1.3": 218 229 version: 0.1.3 219 230 resolution: "@atproto/did@npm:0.1.3" ··· 5575 5586 languageName: node 5576 5587 linkType: hard 5577 5588 5589 + "@noble/curves@npm:^1.7.0": 5590 + version: 1.8.0 5591 + resolution: "@noble/curves@npm:1.8.0" 5592 + dependencies: 5593 + "@noble/hashes": "npm:1.7.0" 5594 + checksum: 10/c54ce84cf54b8bda1a37a10dfae2e49e5b6cdf5dd98b399efa8b8a80a286b3f8f27bde53202cb308353bfd98719938991a78bed6e43f81f13b17f8181b7b82eb 5595 + languageName: node 5596 + linkType: hard 5597 + 5578 5598 "@noble/hashes@npm:1.4.0, @noble/hashes@npm:^1.3.1, @noble/hashes@npm:^1.4.0, @noble/hashes@npm:~1.4.0": 5579 5599 version: 1.4.0 5580 5600 resolution: "@noble/hashes@npm:1.4.0" ··· 5593 5613 version: 1.6.0 5594 5614 resolution: "@noble/hashes@npm:1.6.0" 5595 5615 checksum: 10/b44b043b02adbecd33596adeed97d9f9864c24a2410f7ac3b847986c2ecf1f6f0df76024b3f1b14d6ea954932960d88898fe551fb9d39844a8b870e9f9044ea1 5616 + languageName: node 5617 + linkType: hard 5618 + 5619 + "@noble/hashes@npm:1.7.0, @noble/hashes@npm:^1.6.1": 5620 + version: 1.7.0 5621 + resolution: "@noble/hashes@npm:1.7.0" 5622 + checksum: 10/ab038a816c8c9bb986e92797e3d9c5a5b37c020e0c3edc55bcae5061dbdd457f1f0a22787f83f4787c17415ba0282a20a1e455d36ed0cdcace4ce21ef1869f60 5596 5623 languageName: node 5597 5624 linkType: hard 5598 5625 ··· 9929 9956 languageName: node 9930 9957 linkType: hard 9931 9958 9959 + "@types/qrcode@npm:^1": 9960 + version: 1.5.5 9961 + resolution: "@types/qrcode@npm:1.5.5" 9962 + dependencies: 9963 + "@types/node": "npm:*" 9964 + checksum: 10/a25686339bd2718e6a93943e7807ed68dd9c74a9da28aa77212086ee0ce9a173c0a232af9e3f6835acd09938dfc8a0f98c6bccf1a6c6a905fb003ab07f9e08f2 9965 + languageName: node 9966 + linkType: hard 9967 + 9932 9968 "@types/qs@npm:*": 9933 9969 version: 6.9.15 9934 9970 resolution: "@types/qs@npm:6.9.15" ··· 10855 10891 languageName: node 10856 10892 linkType: hard 10857 10893 10894 + "@zxing/browser@npm:^0.1.5": 10895 + version: 0.1.5 10896 + resolution: "@zxing/browser@npm:0.1.5" 10897 + dependencies: 10898 + "@zxing/text-encoding": "npm:^0.9.0" 10899 + peerDependencies: 10900 + "@zxing/library": ^0.21.0 10901 + dependenciesMeta: 10902 + "@zxing/text-encoding": 10903 + optional: true 10904 + checksum: 10/13df1471dc93b1a54c11df00caebe529fb5e4435d5876e606da048697a039098f5650724710e92cef2db746dbf89b72d04791fd89c6b9a6cf1c1b841dc24cb32 10905 + languageName: node 10906 + linkType: hard 10907 + 10908 + "@zxing/library@npm:^0.21.3": 10909 + version: 0.21.3 10910 + resolution: "@zxing/library@npm:0.21.3" 10911 + dependencies: 10912 + "@zxing/text-encoding": "npm:~0.9.0" 10913 + ts-custom-error: "npm:^3.2.1" 10914 + dependenciesMeta: 10915 + "@zxing/text-encoding": 10916 + optional: true 10917 + checksum: 10/867e7a9be38ea8636050ab5852440e012525c01cd8791448afd15b8ec3ca41de6594202e3f8c90b2d3677b381a6a186eab79e5a431f23d09eb7887cdec772772 10918 + languageName: node 10919 + linkType: hard 10920 + 10921 + "@zxing/text-encoding@npm:^0.9.0, @zxing/text-encoding@npm:~0.9.0": 10922 + version: 0.9.0 10923 + resolution: "@zxing/text-encoding@npm:0.9.0" 10924 + checksum: 10/268e4ef64b8eaa32b990240bdfd1f7b3e2b501a6ed866a565f7c9747f04ac884fbe0537fe12bb05d9241b98fb111270c0fd0023ef0a02d23a6619b4589e98f6b 10925 + languageName: node 10926 + linkType: hard 10927 + 10858 10928 "JSONStream@npm:^1.3.5": 10859 10929 version: 1.3.5 10860 10930 resolution: "JSONStream@npm:1.3.5" ··· 11255 11325 version: 0.0.0-use.local 11256 11326 resolution: "aquareum-desktop@workspace:js/desktop" 11257 11327 dependencies: 11328 + "@atproto/crypto": "npm:^0.4.3" 11258 11329 "@electron-forge/cli": "npm:^7.5.0" 11259 11330 "@electron-forge/maker-deb": "npm:^7.5.0" 11260 11331 "@electron-forge/maker-dmg": "npm:^7.5.0" ··· 11338 11409 "@tamagui/toast": "npm:^1.116.12" 11339 11410 "@tanstack/react-query": "npm:^5.59.19" 11340 11411 "@types/babel__plugin-transform-runtime": "npm:^7" 11412 + "@types/qrcode": "npm:^1" 11341 11413 "@types/react": "npm:~18.3.12" 11342 11414 "@types/uuid": "npm:^10.0.0" 11415 + "@zxing/browser": "npm:^0.1.5" 11416 + "@zxing/library": "npm:^0.21.3" 11343 11417 abortcontroller-polyfill: "npm:^1.7.6" 11344 11418 babel-preset-expo: "npm:~12.0.0" 11345 11419 burnt: "npm:^0.12.2" ··· 11362 11436 hls.js: "npm:^1.5.17" 11363 11437 jose: "npm:^5.9.6" 11364 11438 multiformats: "npm:^13.3.1" 11439 + qrcode: "npm:^1.5.4" 11440 + quietjs-bundle: "npm:^0.1.3" 11365 11441 react: "npm:18.3.1" 11366 11442 react-dom: "npm:18.3.1" 11367 11443 react-native: "npm:0.76.2" ··· 11943 12019 languageName: node 11944 12020 linkType: hard 11945 12021 12022 + "base64-arraybuffer@npm:^1.0.2": 12023 + version: 1.0.2 12024 + resolution: "base64-arraybuffer@npm:1.0.2" 12025 + checksum: 10/15e6400d2d028bf18be4ed97702b11418f8f8779fb8c743251c863b726638d52f69571d4cc1843224da7838abef0949c670bde46936663c45ad078e89fee5c62 12026 + languageName: node 12027 + linkType: hard 12028 + 11946 12029 "base64-js@npm:1.5.1, base64-js@npm:^1.0.2, base64-js@npm:^1.2.3, base64-js@npm:^1.3.0, base64-js@npm:^1.3.1, base64-js@npm:^1.5.1": 11947 12030 version: 1.5.1 11948 12031 resolution: "base64-js@npm:1.5.1" ··· 12568 12651 languageName: node 12569 12652 linkType: hard 12570 12653 12571 - "chalk@npm:^2.0.1, chalk@npm:^2.4.2": 12654 + "chalk@npm:^2.0.1, chalk@npm:^2.4.1, chalk@npm:^2.4.2": 12572 12655 version: 2.4.2 12573 12656 resolution: "chalk@npm:2.4.2" 12574 12657 dependencies: ··· 12596 12679 languageName: node 12597 12680 linkType: hard 12598 12681 12682 + "chalk@npm:^5.3.0": 12683 + version: 5.4.1 12684 + resolution: "chalk@npm:5.4.1" 12685 + checksum: 10/29df3ffcdf25656fed6e95962e2ef86d14dfe03cd50e7074b06bad9ffbbf6089adbb40f75c00744d843685c8d008adaf3aed31476780312553caf07fa86e5bc7 12686 + languageName: node 12687 + linkType: hard 12688 + 12599 12689 "chardet@npm:^0.7.0": 12600 12690 version: 0.7.0 12601 12691 resolution: "chardet@npm:0.7.0" ··· 20309 20399 languageName: node 20310 20400 linkType: hard 20311 20401 20402 + "memorystream@npm:^0.3.1": 20403 + version: 0.3.1 20404 + resolution: "memorystream@npm:0.3.1" 20405 + checksum: 10/2e34a1e35e6eb2e342f788f75f96c16f115b81ff6dd39e6c2f48c78b464dbf5b1a4c6ebfae4c573bd0f8dbe8c57d72bb357c60523be184655260d25855c03902 20406 + languageName: node 20407 + linkType: hard 20408 + 20312 20409 "meow@npm:^8.1.2": 20313 20410 version: 8.1.2 20314 20411 resolution: "meow@npm:8.1.2" ··· 21766 21863 languageName: node 21767 21864 linkType: hard 21768 21865 21866 + "npm-run-all@npm:^4.1.5": 21867 + version: 4.1.5 21868 + resolution: "npm-run-all@npm:4.1.5" 21869 + dependencies: 21870 + ansi-styles: "npm:^3.2.1" 21871 + chalk: "npm:^2.4.1" 21872 + cross-spawn: "npm:^6.0.5" 21873 + memorystream: "npm:^0.3.1" 21874 + minimatch: "npm:^3.0.4" 21875 + pidtree: "npm:^0.3.0" 21876 + read-pkg: "npm:^3.0.0" 21877 + shell-quote: "npm:^1.6.1" 21878 + string.prototype.padend: "npm:^3.0.0" 21879 + bin: 21880 + npm-run-all: bin/npm-run-all/index.js 21881 + run-p: bin/run-p/index.js 21882 + run-s: bin/run-s/index.js 21883 + checksum: 10/46020e92813223d015f4178cce5a2338164be5f25b0c391e256c0e84ac082544986c220013f1be7f002dcac07b81c7ee0cb5c5c30b84fd6ebb6de96a8d713745 21884 + languageName: node 21885 + linkType: hard 21886 + 21769 21887 "npm-run-path@npm:^2.0.0": 21770 21888 version: 2.0.2 21771 21889 resolution: "npm-run-path@npm:2.0.2" ··· 22729 22847 languageName: node 22730 22848 linkType: hard 22731 22849 22850 + "pidtree@npm:^0.3.0": 22851 + version: 0.3.1 22852 + resolution: "pidtree@npm:0.3.1" 22853 + bin: 22854 + pidtree: bin/pidtree.js 22855 + checksum: 10/eb85b841cd168151bfadb984f9514d67a884d6962d4a2d250d4e8acf85cf031d7dab080f7272fb2735f9033364e5058c73eeebbee3cf6fd829169a75d19f189a 22856 + languageName: node 22857 + linkType: hard 22858 + 22732 22859 "pidtree@npm:~0.6.0": 22733 22860 version: 0.6.0 22734 22861 resolution: "pidtree@npm:0.6.0" ··· 23317 23444 languageName: node 23318 23445 linkType: hard 23319 23446 23320 - "qrcode@npm:1.5.4": 23447 + "qrcode@npm:1.5.4, qrcode@npm:^1.5.4": 23321 23448 version: 1.5.4 23322 23449 resolution: "qrcode@npm:1.5.4" 23323 23450 dependencies: ··· 23388 23515 languageName: node 23389 23516 linkType: hard 23390 23517 23518 + "quietjs-bundle@npm:^0.1.3": 23519 + version: 0.1.3 23520 + resolution: "quietjs-bundle@npm:0.1.3" 23521 + dependencies: 23522 + base64-arraybuffer: "npm:^1.0.2" 23523 + chalk: "npm:^5.3.0" 23524 + npm-run-all: "npm:^4.1.5" 23525 + uglify-js: "npm:^3.17.4" 23526 + checksum: 10/319acbfb1ea11798866586b484106e68e99eb7bad15980b37f26c3a1060a890b541d01c31adc0ba0510b2445377b767876053978cb39d1a3b3a76d6eb173ecdc 23527 + languageName: node 23528 + linkType: hard 23529 + 23391 23530 "r-json@npm:^1.2.10": 23392 23531 version: 1.3.0 23393 23532 resolution: "r-json@npm:1.3.0" ··· 25487 25626 languageName: node 25488 25627 linkType: hard 25489 25628 25629 + "string.prototype.padend@npm:^3.0.0": 25630 + version: 3.1.6 25631 + resolution: "string.prototype.padend@npm:3.1.6" 25632 + dependencies: 25633 + call-bind: "npm:^1.0.7" 25634 + define-properties: "npm:^1.2.1" 25635 + es-abstract: "npm:^1.23.2" 25636 + es-object-atoms: "npm:^1.0.0" 25637 + checksum: 10/52cebc58a0252ef45dd0fec3ee4e8655bcc8b6c07b4956c5965542316f5ab3a38ca8d1d06e9804979828fba9de61e59294fe23f64e5d413ac40963a4d4969c19 25638 + languageName: node 25639 + linkType: hard 25640 + 25490 25641 "string.prototype.trim@npm:^1.2.9": 25491 25642 version: 1.2.9 25492 25643 resolution: "string.prototype.trim@npm:1.2.9" ··· 26253 26404 languageName: node 26254 26405 linkType: hard 26255 26406 26407 + "ts-custom-error@npm:^3.2.1": 26408 + version: 3.3.1 26409 + resolution: "ts-custom-error@npm:3.3.1" 26410 + checksum: 10/92e3a2c426bf6049579aeb889b6f9787e0cfb6bb715a1457e2571708be7fe739662ca9eb2a8c61b72a2d32189645f4fbcf1a370087e030d922e9e2a7b7c1c994 26411 + languageName: node 26412 + linkType: hard 26413 + 26256 26414 "ts-interface-checker@npm:^0.1.9": 26257 26415 version: 0.1.13 26258 26416 resolution: "ts-interface-checker@npm:0.1.13" ··· 26622 26780 bin: 26623 26781 uglifyjs: bin/uglifyjs 26624 26782 checksum: 10/44b37f88805565ba478665f4d5560388a072b314c38708046a5b97ca49ec40cb0d34414daff77d44695991098b7596536847e7d87b4590f457fc757e1d2904cc 26783 + languageName: node 26784 + linkType: hard 26785 + 26786 + "uglify-js@npm:^3.17.4": 26787 + version: 3.19.3 26788 + resolution: "uglify-js@npm:3.19.3" 26789 + bin: 26790 + uglifyjs: bin/uglifyjs 26791 + checksum: 10/6b9639c1985d24580b01bb0ab68e78de310d38eeba7db45bec7850ab4093d8ee464d80ccfaceda9c68d1c366efbee28573b52f95e69ac792354c145acd380b11 26625 26792 languageName: node 26626 26793 linkType: hard 26627 26794