Development

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"
   }
 }

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 ---------------------------- */
 

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'),
     ]
   }
 }

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,
+    }
+  }
+}

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()
+  }
+}

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,
+  })
+}

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)
     })
   })