diff --git a/package.json b/package.json index 82fd424..9351f30 100644 --- a/package.json +++ b/package.json @@ -4,12 +4,12 @@ "description": "FFmpeg streaming service", "main": "index.js", "scripts": { - "dev:server": "node dev.mjs | bunyan", - "dev:server:watch": "npm-watch dev:server", + "server:bunyan": "node dev.mjs | bunyan", + "dev": "npm-watch server:bunyan", "test": "echo \"Error: no test specified\" && exit 1" }, "watch": { - "dev:server": { + "server:bunyan": { "patterns": [ "server/*" ], @@ -28,7 +28,7 @@ "service-core": "^3.0.0-beta.17" }, "dependencies": { - "flaska": "^1.2.3", + "flaska": "^1.2.4", "ws": "^8.6.0" } } diff --git a/public/assets/ErbosDraco.woff2 b/public/assets/ErbosDraco.woff2 new file mode 100644 index 0000000..17fa311 Binary files /dev/null and b/public/assets/ErbosDraco.woff2 differ diff --git a/public/assets/app.css b/public/assets/app.css index 2b2ba18..cf166bf 100644 --- a/public/assets/app.css +++ b/public/assets/app.css @@ -20,6 +20,13 @@ src: url("Inter.var.woff2?v=3.19") format("woff2"); } +@font-face { + font-family: 'ErbosDraco'; + font-weight: 400; + font-display: swap; + src: url(data:font/woff2;base64,d09GMgABAAAAAAYEAAoAAAAAatAAAAW1AAEAAAAAAAAAAAAAAAAAAAAAAAAAAAAABmAARAqByzTVewE2AiQDgUoLgUoABCAFgxgHIBuaLhAsCuymh8NM0d9+u0fTqsJgwfM0l72ZZPFnAluUI0zL8gAwOcBsAZMiuSN0pwxJYCN6QrPbAOD53L2/RIJuO/QEDj32QMWr4AIgm+CFCyARfdRDVav/WQaYQIPwz/h/GIly/yq47a0QjvBENLk1y+vZ+f82CocxKBzOEQjFLSxCYhRGYZxk39dZzfjqMk9mOlcT9EBNzRaHCW5jtm6Nj0Drk3lKAVejAYA/W0OfHwAuALAFiqf3HroXzrEH2+VsufYdfwR6aA0BC0ADDtCMgwKacekGZcpoYMNiq9EAbG97Go3Gz+0j24fG/4wYAPAF3EGil9BcBABCAACKuIREGKMpUaJMxaWAS4EmqgE+PhERHhERZcqUqJaothDSjMn56dSIEIQAoYmIkIAmagQIAaYJUw0QCggFCmh8QqohPh4+pkzFo4yPh4eHUDHGoxJQ8TCllusF9L8PXNzG2+oRYlBWi5S0SJJ0RzWaWnt1ey/G1LQXY0vbAoNgZDEbk5JlhljjAOsqskoTQI6Bm3xEmWQvcWkrSY2zq7UqWogQAAKpmDczXttkzC4TdC8GD2Qh5D3L/I0mxJJWM5JiFFywOVHoiHcFsCQLKY1JU5kvMlhjMeKGSNvDUGr12UeWjMRIdq9EVN0AR/K0ZQLyGYuRTnFuFcMiNYv1KV2LhJU4ArTjWtoGJGqW3Y+Q+OGCAEFULy7M+jDitCVtntsmtC/e2eP8gUbonqPDccoJMdoWRjcN0gjJpwrRLC0O34UGy4oT4Ec0qvaJYoG60fCj5nt06w0lFMHhpALyoag+iiLrR89AcyiLYp/Eil3O9x8BW+rOWdHLrcK5oY7eFE6RB+MkV3ipUA2J+d+VmJEtcKSy/4lqVBjE7xzm9z3+zhk9znpeuPGogomG7/PRm2BMA4U3lsrET7PPwz8XfSlwoEyrdThrHeEQ3o5JSZcknqrE1LvpOkI5u+c/q0fbzuShJD2uqinhWIQ1SRMruIIzqt5soGU1i+EDqVH3yyLDc/v1bd4rCd97w3NVO8bzPJDmO7k5PAe4uJv9919l4OUzCCq6+865zsxycur+ITOGyScOepH2915/YXdriUt6d3Q51B3UZa+67MFyinxCHuTjLK3Tbv0gv4aLI4inXHGkfGaF7ENhKh3qS6EyId0V4+0XgNV/gNHmDdbasHF6eer28TrAnETmCuGB/twGiod5bhQo3yOpp6Ix2EfzacAKdzob/NGcbiMrkXc+t1I4aQa/gIPx4lHtlygVomx+X7k/r6vcRrlhXhn6fPzd67/gJ1s6zzTOg5FcDD9HEL04g+LIDUey9gRPDQnz7wN2myc+pN0/6EC/aN5t4MTBIw22YRGHULMSfNpB+OI1uFpG0S/V9pU7xvTYaF+28x8vxda3atTb36i+Ea+tQ+2wqoLaNnsaDb6Pb2q8Oqyo4Xa/IUr3N8TfTG0A6O3zlweeRJWdPSB7t3qHWNvXtP/dko/8ll9fuAielqNg9eyv9azivfF1HvLNfX75sxft4vuPNH2y+y6L+/DW+7dk++uR1639Ye5fKabgeGIuw7jqEyflWfn7TWPmM9SQHEKHwYf1F3EFzS6bC+fdpaKZWZXqXylbuFCJaFcIIRn5XxDib8QtduOuPZg5/jcVu/DfPLa8XrurwK9h+AcPbgY7yrgA/Hg1+eBv0NlyfgFlNAAKXACcO8BLZwtF2e37COW1VG6oh2zy1A39sM6KHbtjH4XCXVF3VMsl78uggAEXtACgaXyoxBUOyq1AP7pWim79K43PwMrhkJmVyx5erwq08v//okW647NfPmXZjNFhI+ZYp9yeM2gdcXvChMFlVq9uPXodnJHKZ1kHzLhdl1s9Zs2xTk8ZzFin7HPGOYPD5ifcnnFInplz3pwZ83VzLKGtO7NF/dzY4IxZo3nG6tHZfbexo3Zc66gdTx3QbjakDwAAAA==); +} + html { box-sizing: border-box; } @@ -168,6 +175,14 @@ main { flex: 2 1 auto; } +.text { + font-family: ErbosDraco; + font-size: 16px; + padding: 1rem; + background: black; + color: white; +} + /* ---------------------------- Extra ---------------------------- */ diff --git a/public/assets/app.js b/public/assets/app.js index 1baae9a..d72fe96 100644 --- a/public/assets/app.js +++ b/public/assets/app.js @@ -1,3 +1,25 @@ + +class Mitt { + constructor() { + this.all = new Map() + } + on(type, handler) { + var group = this.all.get(type) + group ? group.push(handler) : this.all.set(type, [handler]) + } + off(type, handler) { + var group = this.all.get(type) + group && (handler ? group.splice(group.indexOf(handler) >>> 0, 1) : group.set(type, [])) + } + emit(type, payload) { + var group = this.all.get(type) + if (group) { group.forEach(x => x(payload)) } + if (group = this.all.get('*')) { + group.forEach(x => x(type, payload)) + } + } +} + class Animator { constructor() { this.running = null @@ -23,9 +45,10 @@ class Animator { const animator = new Animator() -class WSClient +class WSClient extends Mitt { constructor(url = 'ws://localhost:4040') { + super() this.url = url this.socket = null this.connected = false @@ -64,13 +87,16 @@ class WSClient this.connectingAt = null this.reconnectingAt = null this.connected = true - this.send({ msg: 'Hello Server!' }) + // this.send('onopen', { msg: 'Hello Server!' }) m.redraw() } - send(payload) { + send(type, payload) { if (!this.connected) return - this.socket.send(JSON.stringify(payload)) + this.socket.send(JSON.stringify({ + type: type, + payload: payload, + })) } onerror() { @@ -102,16 +128,22 @@ class WSClient onmessage(event) { try { let data = JSON.parse(event.data) - console.log('got message', data) + this.emit(data.type, data.payload) } catch (err) { console.error(err) } } } const client = new WSClient() +let encodingStatus = {} + +client.on('status', function(newStatus) { + encodingStatus = newStatus + m.redraw() +}) + class Header { view(vnode) { - console.log(client.connecting) return [ m('h1', 'Fíladelfíu streymi'), m('div.status.green', { hidden: true }, 'No errors'), @@ -125,7 +157,9 @@ class Header { : this.viewWaitingCircle(), ]), m('span', 'Live:'), - m('div.led'), + m('div.led', { + class: encodingStatus.encoding ? 'led-green' : '', + }), ] } viewConnectingBar() { @@ -233,7 +267,8 @@ class Nav { class Status { view(vnode) { return [ - m('span', 'Hello world') + m('div.text', 'Some text here'), + m('span', 'Hello world'), ] } } diff --git a/public/assets/favicon - Copy.ico b/public/assets/favicon - Copy.ico new file mode 100644 index 0000000..7823b8b Binary files /dev/null and b/public/assets/favicon - Copy.ico differ diff --git a/public/assets/favicon.ico b/public/assets/favicon.ico index 7823b8b..b17826d 100644 Binary files a/public/assets/favicon.ico and b/public/assets/favicon.ico differ diff --git a/server/ffmpeg/encoder.mjs b/server/ffmpeg/encoder.mjs new file mode 100644 index 0000000..1ca4c90 --- /dev/null +++ b/server/ffmpeg/encoder.mjs @@ -0,0 +1,65 @@ +import os from 'os' +import { EventEmitter } from 'events' +import { spawn, execSync } from 'child_process' + +export default class Encoder extends EventEmitter { + constructor(opts = {}) { + super() + Object.assign(this, { + encoding: false, + program: 'ffmpeg.exe', + options: ['--help'], + processor: null, + }) + } + + start() { + if (this.processor) return + + this.processor = spawn(this.program, this.options, { + shell: true, + }) + + this.encoding = true + + this.emit('status', this.status()) + + this.processor.stdout.on('data', (data) => { + this.emit('stdout', data.toString()) + }) + this.processor.stderr.on('data', (data) => { + this.emit('stderr', data.toString()) + }) + + this.processor.on('error', (err) => { + this.processor = null + this.encoding = false + this.emit('error', err) + this.emit('status', this.status()) + }) + this.processor.on('exit', (code) => { + this.processor = null + this.encoding = false + this.emit('exit', { code: code }) + this.emit('status', this.status()) + }) + } + + stop() { + if(os.platform() === 'win32'){ + try { + execSync('taskkill /pid ' + this.processor.pid + ' /T /F') + } catch {} + } else { + this.processor.kill(); + } + this.encoding = false + this.processor = null + } + + status() { + return { + encoding: this.encoding, + } + } +} \ No newline at end of file diff --git a/server/ffmpeg/monitor.mjs b/server/ffmpeg/monitor.mjs new file mode 100644 index 0000000..3544561 --- /dev/null +++ b/server/ffmpeg/monitor.mjs @@ -0,0 +1,27 @@ +import { EventEmitter } from 'events' + +export default class Monitor extends EventEmitter { + constructor(wss, db, encoder, opts = {}) { + super() + this.wss = wss + this.db = db + this.encoder = encoder + this.encoder.on('stdout', (data) => { + console.log('stdout', { data }) + }) + this.encoder.on('stderr', (data) => { + console.log('stderr', { data }) + }) + this.encoder.on('status', (status) => { + this.emit('status', status) + }) + } + + start() { + this.encoder.start() + } + + status() { + return this.encoder.status() + } +} \ No newline at end of file diff --git a/server/ffmpeg/startup.mjs b/server/ffmpeg/startup.mjs new file mode 100644 index 0000000..52e587c --- /dev/null +++ b/server/ffmpeg/startup.mjs @@ -0,0 +1,12 @@ +import Encoder from './encoder.mjs' + +export function startup(core) { + core.db.data.ffmpeg = core.db.data.ffmpeg || {} + core.db.data.ffmpeg.program = 'ffmpeg.exe' + core.db.data.ffmpeg.options = ['--help'] + core.db.write() + return new Encoder({ + program: core.db.data.ffmpeg.program, + options: core.db.data.ffmpeg.options, + }) +} \ No newline at end of file diff --git a/server/server.mjs b/server/server.mjs index 37381fd..1b59c90 100644 --- a/server/server.mjs +++ b/server/server.mjs @@ -1,6 +1,9 @@ +import crypto from 'crypto' import { Flaska, QueryHandler } from 'flaska' import { WebSocket, WebSocketServer } from 'ws' +import { startup } from './ffmpeg/startup.mjs' +import Monitor from './ffmpeg/monitor.mjs' import ServeHandler from './serve.mjs' export function run(http, port, core) { @@ -46,12 +49,34 @@ export function run(http, port, core) { const wss = new WebSocketServer({ server: flaska.server }) + const encoder = startup(core) + const monitor = new Monitor(wss, core.db, encoder) + monitor.start() + + monitor.on('status', (status) => { + wss.clients.forEach(function each(client) { + if (!client.readyState === WebSocket.OPEN) { + console.log('closed', client) + } + client.sendevent('status', status); + }) + }) + wss.on('connection', function(ws) { - console.log('new connection') + ws.id = crypto.randomBytes(6).toString('base64') ws.isAlive = true + core.log.info({ id: ws.id, ip: ws._socket.remoteAddress }, 'New connection') ws.on('pong', function() { ws.isAlive = true }) + ws.sendevent = function(type, data) { + ws.send(JSON.stringify({ + type: type, + payload: data, + })) + } + + ws.sendevent('status', monitor.status()) ws.on('message', function message(data, isBinary) { if (isBinary) { @@ -64,19 +89,20 @@ export function run(http, port, core) { core.log.error(err) return } - console.log('got', payload) - wss.clients.forEach(function each(client) { + if (!payload.type || typeof(payload.type) !== 'string') { + core.log.error(new Error('Got payload but it was missing type: ' + data.toString())) + return + } + console.log('got', payload.type, payload.payload) + /*wss.clients.forEach(function each(client) { if (client.readyState === WebSocket.OPEN) { client.send(JSON.stringify(payload)); } - }) + })*/ }) }) const interval = setInterval(function() { - if (wss.clients.length > 0) { - core.log.info('Connected clients: ' + wss.clients.length) - } wss.clients.forEach(function(ws) { if (ws.isAlive === false) { return ws.terminate() @@ -85,9 +111,10 @@ export function run(http, port, core) { ws.isAlive = false ws.ping() }) - }, 5000) + }, 15000) wss.on('close', function() { + console.log('closing') clearInterval(interval) }) })