Status updates

README.md

@@ -5,7 +5,7 @@ This repository actually holds three programs, so be careful to not lost in the
 * After `make` completes, `wsrA` is the relay program, which can be run as so: `./wsrA port=12345 key=MyFancyStreamingKeyIsOVERHERE`.
 * Using the above example, one should stream to the HTTP path `/push/MyFancyStreamingKeyIsOVERHERE`. All other paths are expected to be used by WebSocket clients.
 * If using a reverse proxy (highly recommended), make sure to disable request buffering and the maximum request size. For nginx, you want `proxy_request_buffering off; client_max_body_size 0;`.
-* `index.html`, `blarf.js`, `blarfwork.js`, `support.js` and `support.wasm` are the frontend, which must be accessible to the browser.
+* `index.html`, `blarf.js`, `blarfwork.js`, `support.js`, `support.wasm` and `rawpcmworklet.js` are the frontend, which must be accessible to the browser.
 * You may insert a file named `intermission.jpg` that is shown when the stream is offline.
 * Of course you'll have to change the feed and chat endpoints in the `index.html`, if you don't want my stream.
 * You may also disable chat by setting `ENABLE_CHAT` to `false`.

blarf.js

@@ -2,22 +2,51 @@
 	var VideoQueue = []
 	var AudioQueue = []
 	
+	class DynamicTypedArray {
+		constructor(type) {
+				this.type = type
+				this.backend = new type(1024)
+				this.length = 0
+		}
+		add(b) {
+			if(this.length + b.length > this.backend.length) {
+					var newlen = this.backend.length
+					while(this.length + b.length > newlen) { newlen = newlen * 2 }
+					var be2 = new this.type(newlen)
+					be2.set(this.backend, 0)
+					this.backend = be2
+			}
+			this.backend.set(b, this.length)
+			this.length += b.length
+		}
+	}
+	
 	var BlarfEl = document.getElementById("BLARF")
 	BlarfEl.innerHTML = `
 		<canvas width="1280" height="720"></canvas>
 		<div class="MKVControls">
-			<div class="MKVSpeaker"><span class="MKVSpeakerOff">🔈&#xFE0E;</span><span class="MKVSpeakerOn" style="display:none;">🔊&#xFE0E;</span></div>
+			<div>
-			<span class="MKVCurrentTime">00:00:00</span>
+				<div class="MKVSpeaker"><span class="MKVSpeakerOff">🔈&#xFE0E;</span><span class="MKVSpeakerOn" style="display:none;">🔊&#xFE0E;</span></div>
-			<span class="MKVStats"></span>
+				<span class="MKVCurrentTime">00:00:00</span>
+				<span class="MKVStats"></span>
+			</div>
+			<div>
+				<span class="MKVStatus"></span>
+			</div>
 		</div>
 	`
 	
 	var Canvus = BlarfEl.querySelector("canvas")
 	var CanvCtx = Canvus.getContext("2d")
+	var CanvImageData
+	
+	var LatencyMS = 1000
 	
 	var AudCtx
 	var AudScript, AudWorklet
 	var AudHz
+	var AudMuted = true
+	var AudSampleIndex = 0
 	
 	function create_audio(hz, channels) {
 		if(AudCtx) {
@@ -52,8 +81,10 @@
 				while(AudioQueue.length && leftToWrite) {
 					var amount = Math.min(leftToWrite, AudioQueue[0].left.length)
 					
-					outL.set(AudioQueue[0].left.subarray(0, amount), offset)
+					if(!AudMuted) {
-					if(outR) outR.set(AudioQueue[0].right.subarray(0, amount), offset)
+						outL.set(AudioQueue[0].left.subarray(0, amount), offset)
+						if(outR) outR.set(AudioQueue[0].right.subarray(0, amount), offset)
+					}
 					
 					AudioQueue[0].left = AudioQueue[0].left.subarray(amount)
 					if(outR) AudioQueue[0].right = AudioQueue[0].right.subarray(amount)
@@ -77,11 +108,16 @@
 	interruptcontrols()
 
 	function togglemute() {
-		if(AudCtx)
+		if(!AudCtx) {
-		if(document.querySelector(".MKVSpeakerOn").style.display == "none") {
+			return;
-			AudCtx.resume()
+		}
-		} else {
+		
-			AudCtx.suspend()
+		AudCtx.resume()
+		
+		AudMuted = !AudMuted
+		
+		if(AudWorklet) {
+			AudWorklet.port.postMessage(AudMuted)
 		}
 		
 		document.querySelectorAll(".MKVSpeaker *").forEach(function(el) { el.style.display = el.style.display == "none" ? "" : "none" })
@@ -92,7 +128,7 @@
 	document.querySelector(".MKVSpeaker").onclick = togglemute
 
 	document.onkeypress = function(e) {
-		if(e.key.toUpperCase() == "M") {
+		if(document.activeElement.tagName != "TEXTAREA" && e.key.toUpperCase() == "M") {
 			togglemute()
 		}
 	}
@@ -118,32 +154,31 @@
 		}
 	}
 	
+	var BufferPool = new Set()
+	
 	var Statistics = {}
 	var TheWorker = new Worker("blarfwork.js")
 	TheWorker.onmessage = function(e) {
 		if(e.data.width) {
-			var imgData = new ImageData(new Uint8ClampedArray(e.data.data.buffer), e.data.width, e.data.height, {colorSpace: "srgb"})
+//			var imgData = new ImageData(new Uint8ClampedArray(e.data.data.buffer), e.data.width, e.data.height, {colorSpace: "srgb"})
-			VideoQueue.push({t: e.data.t, imgData: imgData})
+			var b
+			if(BufferPool.size == 0) {
+				b = new Uint8ClampedArray(e.data.data.buffer)
+			} else {
+				for(const v of BufferPool) {
+					b = v
+					break
+				}
+				BufferPool.delete(b)
+				b.set(e.data.data)
+			}
+			VideoQueue.push({t: e.data.t, imgData: b, w: e.data.width, h: e.data.height})
 		} else if(e.data.samples) {
-			AudioQueue.push({left: e.data.left, right: e.data.right || e.data.left})
+			AudioQueue.push({t: e.data.t, left: e.data.left, right: e.data.right || e.data.left})
 			
-			// Audio may be loaded but it might not play because of autoplay permissions
+			if(AudCtx.state == "running" && AudWorklet && AudioQueue.length) {
-			// In this case the audio queue will fill up and cause ever-increasing AV desync
+				AudWorklet.port.postMessage(merge_audio_queue())
-			
+				AudioQueue.length = 0
-			if(AudCtx && AudCtx.state != "running") {
-				var durationInAudioQueue = AudioQueue.length ? AudioQueue.reduce((acc, el) => acc + el.left.length, 0) : 0
-				
-				//if(AudWorklet) {
-				//	// With audio worklets, crop to 1024 samples max to prevent ring buffer overflow
-				//	
-				//	crop_audio_queue(Math.max(durationInAudioQueue - 1024, 0))
-				//} else {
-				//	// Without audio worklets we use a FIFO and can crop to the duration in the video queue
-				//	
-					var durationToRemove = Math.max(durationInAudioQueue - (VideoQueue.length ? (VideoQueue[VideoQueue.length - 1].t - VideoQueue[0].t) : 0) * AudHz / 1000, 0)
-				//	
-					crop_audio_queue(durationToRemove)
-				//}
 			}
 		}
 		
@@ -165,10 +200,6 @@
 		}
 	}
 
-	Canvus.onclick = function() {
-		if(AudCtx) AudCtx.resume()
-	}
-
 	var VideoBufferingOffset = 0
 
 	function toHex(buffer) {
@@ -337,10 +368,7 @@
 				this.currentClusterTime = EBMLParser.vi_to_i(data)
 				
 				if(!RenderStartTime) {
-					RenderStartTime = document.timeline.currentTime + 1000
+					RenderStartTime = performance.now()
-				}
-				if(!VideoStartTime) {
-					VideoStartTime = this.currentClusterTime
 				}
 			} else if(elID == 0xA3) {
 				// Cluster -> SimpleBlock
@@ -357,10 +385,16 @@
 				var TotalTime = (this.currentClusterTime + timestamp) / 1000
 				document.querySelector(".MKVCurrentTime").innerText = pad(Math.floor(TotalTime / 3600), 2) + ":" + pad(Math.floor(TotalTime / 60 % 60), 2) + ":" + pad(Math.floor(TotalTime % 60), 2)
 				
+				var playerTimestamp = this.currentClusterTime + timestamp
+				
 				if(track) {
+					if(!VideoStartTime) {
+						VideoStartTime = playerTimestamp
+					}
+					
 					var packet = data.subarray(4)
 					
-					TheWorker.postMessage({cmd: "decode", id: trackID, t: timestamp + this.currentClusterTime - VideoStartTime, packet: packet, kf: kf})
+					TheWorker.postMessage({cmd: "decode", id: trackID, t: playerTimestamp - VideoStartTime, packet: packet, kf: kf})
 				}
 			}
 		}
@@ -382,6 +416,9 @@
 				if(track.type == "video") {
 					Canvus.width = track.width
 					Canvus.height = track.height
+					CanvImageData = new ImageData(new Uint8ClampedArray(Canvus.width * Canvus.height * 4), Canvus.width, Canvus.height, {"colorSpace": "srgb"})
+					RenderStartTime = null
+					VideoStartTime = null
 				} else {
 					create_audio(track.samplerate, track.channels)
 				}
@@ -414,24 +451,28 @@
 			s += AudioQueue[i].left.length
 		}
 		
-		return {msg: "data", left: L, right: R}
+		var ret = {msg: "data", t: AudSampleIndex, left: L, right: R}
+		
+		AudSampleIndex += L.length
+		
+		return ret
 	}
 
 	function reconnect_ws() {
 		var ws = new WebSocket(BlarfEl.getAttribute("data-target"))
 		ws.binaryType = "arraybuffer"
 		ws.onmessage = function(ev) {
-			ebml.poosh(new Uint8Array(ev.data))
+			if(typeof ev.data === "string") {
-			ebml.parse()
+				var obj = JSON.parse(ev.data)
-			
+				if(obj.status) {
-			// It would make more sense for this to be in `render` but we need the guarantee that this will run when the tab is out of focus
+					BlarfEl.querySelector(".MKVStatus").innerHTML = "&bull; " + obj.status.viewer_count
-			if(AudCtx.state == "running" && AudWorklet && AudioQueue.length) {
+				}
-				AudWorklet.port.postMessage(merge_audio_queue())
+			} else {
-				AudioQueue.length = 0
+				ebml.poosh(new Uint8Array(ev.data))
-			}
+				ebml.parse()
-			if(VideoQueue.length) {
+				
-				while(document.timeline.currentTime - VideoQueue[0].t > 5000) {
+				while(document.hidden && VideoQueue.length > 1 && VideoQueue[VideoQueue.length - 1].t - VideoQueue[0].t <= LatencyMS) {
-					VideoQueue.shift()
+					BufferPool.add(VideoQueue.shift().imgData)
 				}
 			}
 		}
@@ -442,14 +483,37 @@
 	reconnect_ws()
 
 	function render(timestamp) {
-		document.querySelector(".MKVControls").style.opacity = Math.max(0, Math.min(1, 5 - (timestamp - LastControlsInterrupt) / 1000))
+		try {
-		
+			document.querySelector(".MKVControls").style.opacity = Math.max(0, Math.min(1, 5 - (timestamp - LastControlsInterrupt) / 1000))
-		while(RenderStartTime && VideoQueue.length && VideoQueue[0].t + VideoBufferingOffset <= (timestamp - RenderStartTime)) {
+			
-			CanvCtx.putImageData(VideoQueue[0].imgData, 0, 0)
+			var nextImg = null
-			VideoQueue.shift()
+			while(RenderStartTime && VideoQueue.length && VideoQueue[0].t <= (timestamp - RenderStartTime - LatencyMS)) {
+				if(nextImg) BufferPool.add(nextImg.imgData)
+				nextImg = VideoQueue[0]
+				VideoQueue.shift()
+			}
+			
+			if(nextImg) {
+				document.querySelector(".MKVControls").style.display = null
+				
+				// Prevent the audio queue filling up and causing ever-increasing AV desync
+				if(AudCtx && AudCtx.state != "running" && AudioQueue && AudioQueue.length) {
+					if(AudioQueue[0].t < nextImg.t) {
+						crop_audio_queue(Math.round((nextImg.t - AudioQueue[0].t) / 1000 * AudHz))
+					}
+				}
+				
+				CanvImageData.data.set(nextImg.imgData)
+				CanvCtx.putImageData(CanvImageData, 0, 0)
+				BufferPool.add(nextImg.imgData)
+			}
+		} catch(e) {
+			console.error(e)
 		}
 		
 		requestAnimationFrame(render)
 	}
 	requestAnimationFrame(render)
+	
+	document.querySelector(".MKVControls").style.display = "none"
 })()

index.html

@@ -42,6 +42,9 @@
 				font-size: 0.4cm;
 				background: rgb(0, 0, 0);
 				background: linear-gradient(0deg, rgba(0, 0, 0, 1) 0%, rgba(0, 0, 0, 0) 100%);
+				display: flex;
+				justify-content: space-between;
+				align-items: baseline;
 			}
 			div#BLARF .MKVControls > * {
 				vertical-align: middle;
@@ -53,6 +56,9 @@
 				cursor: pointer;
 				font-size: 0.75cm;
 			}
+			div#BLARF .MKVStatus {
+				margin-right: 0.5em;
+			}
 			div#BLARF > canvas {
 				background: url(intermission.jpg) black;
 				background-position: 0 30%;
@@ -71,6 +77,10 @@
 				display: block;
 				line-height: initial;
 			}
+			span.chat-msg__heading {
+				width: inherit !important;
+				margin-bottom: 0;
+			}
 			
 			@media(max-aspect-ratio: 1) {
 				div.everything {
@@ -131,10 +141,9 @@
 				converse.initialize({
 					view_mode: 'embedded',
 					websocket_url: CHAT_HOST_WS_URL,
-					login: 'anonymous',
+					authentication: 'anonymous',
-					jid: un + '@' + CHAT_HOST,
+					jid: CHAT_HOST,
 					auto_login: true,
-					password: 'lol',
 					auto_join_rooms: [CHAT_MUC],
 					show_message_avatar: false,
 					show_controlbox_by_default: false,

main2.c

@@ -10,6 +10,7 @@
 #include<sys/types.h>
 #include<fcntl.h>
 #include<unistd.h>
+#include<time.h>
 #endif
 
 #include<stdbool.h>
@@ -123,13 +124,14 @@ static void transmit_all(const char *buf, size_t sz) {
 	}
 }
 
+#define WS_TXT 1
 #define WS_BIN 2
 #define WS_CLOSE 8
 #define WS_FIN 128
 #define WS_HEADER_MAX 10
-static int ws_header(size_t sz, uint8_t hdr[static WS_HEADER_MAX]) {
+static int ws_header(size_t sz, bool binary, uint8_t hdr[static WS_HEADER_MAX]) {
 	int i;
-	hdr[0] = WS_BIN | WS_FIN;
+	hdr[0] = (binary ? WS_BIN : WS_TXT) | WS_FIN;
 	if(sz < 126) {
 		hdr[1] = sz;
 		i = 2;
@@ -154,27 +156,53 @@ static int ws_header(size_t sz, uint8_t hdr[static WS_HEADER_MAX]) {
 	return i;
 }
 
-static void ws_send(Client *cli, const uint8_t *buf, size_t sz) {
+static void ws_send(Client *cli, bool binary, const uint8_t *buf, size_t sz) {
 	if(sz == 0) return;
 	
 	uint8_t wshdr[WS_HEADER_MAX];
-	int wshdrsz = ws_header(sz, wshdr);
+	int wshdrsz = ws_header(sz, binary, wshdr);
 	
 	transmit(cli, wshdr, wshdrsz);
 	transmit(cli, buf, sz);
 }
 
-static void ws_broadcast(const uint8_t *buf, size_t sz) {
+static void ws_broadcast(bool binary, const uint8_t *buf, size_t sz) {
 	if(sz == 0) return;
 	
 	uint8_t wshdr[WS_HEADER_MAX];
-	int wshdrsz = ws_header(sz, wshdr);
+	int wshdrsz = ws_header(sz, binary, wshdr);
 	
 	transmit_all(wshdr, wshdrsz);
 	transmit_all(buf, sz);
 }
 
+static bool should_send_status_update() {
+	static uint64_t lastSec = 0;
+	
+	struct timespec tv;
+	clock_gettime(CLOCK_MONOTONIC, &tv);
+	
+	if(tv.tv_sec - lastSec < 10) {
+		return false;
+	}
+	
+	lastSec = tv.tv_sec;
+	
+	return true;
+}
+
+static void send_status_update() {
+	char buf[512] = {};
+	snprintf(buf, sizeof(buf) - 1, "{\"status\": {\"viewer_count\": %lu}}", clientsSz);
+	
+	ws_broadcast(false, buf, strlen(buf));
+}
+
 static void stream_step(const uint8_t *newbuf, size_t newsz) {
+	if(should_send_status_update()) {
+		send_status_update();
+	}
+	
 	if(Stream.state == LOADING_HEADER) {
 		Stream.mkvHeader = realloc(Stream.mkvHeader, Stream.mkvHeaderSz + newsz);
 		memcpy(Stream.mkvHeader + Stream.mkvHeaderSz, newbuf, newsz);
@@ -182,16 +210,20 @@ static void stream_step(const uint8_t *newbuf, size_t newsz) {
 		
 		uint8_t *clusterEl = memmem(Stream.mkvHeader, Stream.mkvHeaderSz, "\x1F\x43\xB6\x75", 4);
 		if(clusterEl) {
-			ws_broadcast(Stream.mkvHeader, clusterEl - Stream.mkvHeader);
+			ws_broadcast(true, Stream.mkvHeader, clusterEl - Stream.mkvHeader);
-			ws_broadcast(clusterEl, Stream.mkvHeader + Stream.mkvHeaderSz - clusterEl);
+			ws_broadcast(true, clusterEl, Stream.mkvHeader + Stream.mkvHeaderSz - clusterEl);
 			
 			Stream.mkvHeaderSz = clusterEl - Stream.mkvHeader;
 			Stream.state = STREAMING;
 		}
 	} else {
+		static const uint8_t rootEl[4] = "\x1A\x45\xDF\xA3";
+		
 		int i;
 		for(i = 0; i < newsz; i++) {
-			if(newbuf[i] == "\x1A\x45\xDF\xA3"[Stream.stateChangeIdx]) { 
+			if(newbuf[i] == rootEl[0]) {
+				Stream.stateChangeIdx = 1;
+			} else if(newbuf[i] == rootEl[Stream.stateChangeIdx]) { 
 				Stream.stateChangeIdx++;
 				
 				if(Stream.stateChangeIdx == 4) {
@@ -206,15 +238,17 @@ static void stream_step(const uint8_t *newbuf, size_t newsz) {
 		}
 		
 		if(Stream.state == LOADING_HEADER) {
+			puts("New header");
+			
 			if(i > 4) {
-				ws_broadcast(newbuf, i - 4);
+				ws_broadcast(true, newbuf, i - 4);
 			}
 			
 			Stream.mkvHeader = realloc(Stream.mkvHeader, Stream.mkvHeaderSz = 4 + (newsz - i));
 			memcpy(Stream.mkvHeader, "\x1A\x45\xDF\xA3", 4);
 			memcpy(Stream.mkvHeader + 4, newbuf + i, newsz - i);
 		} else {
-			ws_broadcast(newbuf, newsz);
+			ws_broadcast(true, newbuf, newsz);
 		}
 	}
 }
@@ -298,7 +332,7 @@ static int handle(Client *cli) {
 				
 				if(Stream.state == STREAMING && Stream.mkvHeader) {
 					printf("Sending header\n");
-					ws_send(cli, Stream.mkvHeader, Stream.mkvHeaderSz);
+					ws_send(cli, true, Stream.mkvHeader, Stream.mkvHeaderSz);
 				}
 			}
 			

rawpcmworklet.js

@@ -10,17 +10,23 @@ class RawPCMWorklet extends AudioWorkletProcessor {
 	constructor() {
 		super()
 		
-		this.ringL = new Float32Array(65536)
+		this.ringL = new Float32Array(144000)
-		this.ringR = new Float32Array(65536)
+		this.ringR = new Float32Array(144000)
-		this.ringWrite = 0
 		this.ringRead = 0
+		this.mute = true
 		
 		for(var z = 0; z < 65536; z++) {
 			this.ringL[z] = Math.sin(z / 128 * 2 * Math.PI) * 0.3
 		}
 		
 		this.port.onmessage = (event) => {
+			if(event.data === true || event.data === false) {
+				this.mute = event.data
+				return
+			}
+			
 			var newaudioframes = event.data
+			var writeIndex = newaudioframes.t
 			
 			var newlen = newaudioframes.left.length
 			
@@ -29,22 +35,18 @@ class RawPCMWorklet extends AudioWorkletProcessor {
 				newaudioframes.right = newaudioframes.right.slice(newaudioframes.right.length - this.ringL.length)
 			}
 			
-			if(this.ringWrite % this.ringL.length + newaudioframes.left.length <= this.ringL.length) {
+			if(writeIndex % this.ringL.length + newaudioframes.left.length <= this.ringL.length) {
-				this.ringL.set(newaudioframes.left, this.ringWrite % this.ringL.length)
+				this.ringL.set(newaudioframes.left, writeIndex % this.ringL.length)
-				this.ringR.set(newaudioframes.right, this.ringWrite % this.ringL.length)
+				this.ringR.set(newaudioframes.right, writeIndex % this.ringL.length)
 			} else {
-				var boundary = this.ringL.length - this.ringWrite % this.ringL.length
+				var boundary = this.ringL.length - writeIndex % this.ringL.length
 				
-				this.ringL.set(newaudioframes.left.slice(0, boundary), this.ringWrite % this.ringL.length)
+				this.ringL.set(newaudioframes.left.slice(0, boundary), writeIndex % this.ringL.length)
 				this.ringL.set(newaudioframes.left.slice(boundary), 0)
 				
-				this.ringR.set(newaudioframes.right.slice(0, boundary), this.ringWrite % this.ringL.length)
+				this.ringR.set(newaudioframes.right.slice(0, boundary), writeIndex % this.ringL.length)
 				this.ringR.set(newaudioframes.right.slice(boundary), 0)
 			}
-			
-			this.ringWrite += newlen
-			
-			console.log(this.ringWrite - this.ringRead)
 		}
 	}
 
@@ -58,23 +60,28 @@ class RawPCMWorklet extends AudioWorkletProcessor {
 			return true
 		}*/
 		
-		var available = Math.min(left.length, Math.max(0, this.ringWrite - this.ringRead))
+		//var available = Math.min(left.length, Math.max(0, this.ringWrite - this.ringRead))
+		var available = left.length
 		
-		if(this.ringRead % this.ringL.length + available <= this.ringL.length) {
+		if(this.mute === false) {
-			left.set(this.ringL.slice(this.ringRead % this.ringL.length, this.ringRead % this.ringL.length + available))
+			if(this.ringRead % this.ringL.length + available <= this.ringL.length) {
-			right.set(this.ringR.slice(this.ringRead % this.ringL.length, this.ringRead % this.ringL.length + available))
+				left.set(this.ringL.slice(this.ringRead % this.ringL.length, this.ringRead % this.ringL.length + available))
-		} else {
+				right.set(this.ringR.slice(this.ringRead % this.ringL.length, this.ringRead % this.ringL.length + available))
-			left.set(this.ringL.slice(this.ringRead % this.ringL.length))
+			} else {
-			right.set(this.ringR.slice(this.ringRead % this.ringL.length))
+				left.set(this.ringL.slice(this.ringRead % this.ringL.length))
-			
+				right.set(this.ringR.slice(this.ringRead % this.ringL.length))
-			var boundary = this.ringL.length - this.ringRead % this.ringL.length
+				
-			
+				var boundary = this.ringL.length - this.ringRead % this.ringL.length
-			left.set(this.ringL.slice(0, available - boundary), boundary)
+				
-			right.set(this.ringR.slice(0, available - boundary), boundary)
+				left.set(this.ringL.slice(0, available - boundary), boundary)
+				right.set(this.ringR.slice(0, available - boundary), boundary)
+			}
 		}
 		
 		this.ringRead += left.length
 		
+		//console.log(this.ringRead / 44100)
+		
 		/*for(var s = 0; s < available; s++) {
 			var sw = Math.sin((this.debug + s) / 48000 * 440 * 2 * 3.1415926) * 0.3
 			left[s] = sw