collar-dev-server/index.js

const path = require('path');
const express = require('express');
const app = express();
const http = require('http');

const bodyParser = require('body-parser');

var argv = require('minimist')(process.argv.slice(2));

var port = argv.p || 7500;

if (argv.v) {
  let pack = require('./package.json');
  console.log(`collar-dev-server version ${pack.version}`);
  process.exit();
}

// const io = require('socket.io')(server);

app.use(bodyParser.json());
app.use(bodyParser.urlencoded({ extended: false }));
app.use(express.static(path.join(__dirname, 'public')));

var elementBuffer = [];
var signalBuffer = [];

var apps = {}; // app id

const DEFAULT_ELEMENT_BLOCK = 10;
const DEFAULT_SIGNAL_BLOCK = 10;

var record = false;
var count = 0;

function getProcessId(data) {
  let processId = '__anonymous__';
  if (data && data.process) {
    processId = data.process;
  }
  return processId;
}

const WebSocketServer = require('collar-websocket').Server;

const server = http.createServer(app);
const wss = new WebSocketServer(server);
wss.onAuth((id, secret, done) => {
  // todo: authenticate the requests
  // accept all request
  done(null, {id});
});
wss.on('connect', (socket) => {
  let ns = socket.ns;

  if (ns === '/client') {
    console.log(`client [${socket.clientId}] connected`);
    /*
     * Clients
     */
    socket.on('start record', (data) => {
      console.log('start recording');
      record = true;
      count = new Date().getTime();
      signalBuffer = [];
    });

    socket.on('stop record', (data) => {
      console.log('stop recording');
      record = false;
      count = new Date().getTime();
      signalBuffer = [];
    });

    socket.on('push', (data) => {
      var id = data.nodeId;
      var signal = data.signal;

      wss.of('/app').emit('push', data);
    });

    socket.on('send', (data) => {
      var id = data.nodeId;
      var signal = data.signal;

      wss.of('/app').emit('send', data);
    });

    socket.on('clear', (data) => {
      elementBuffer = [];
      signalBuffer = [];
    });

    socket.on('ready to receive elements', (blockSize) => {
      if (!blockSize || typeof blockSize !== 'number') {
        blockSize = DEFAULT_ELEMENT_BLOCK
      }
      
      let tobesent = [];
      for (let i = 0; i < blockSize; i++) {
        if (elementBuffer.length > i) {
          tobesent.push(elementBuffer.shift());
        } else {
          break;
        }
      }

      if (tobesent.length > 0) {
        wss.of('/client').emit('append elements', {
          elements: tobesent
        });
      }
    });

    socket.on('ready to receive signals', (blockSize) => {
      if (!record) return;
      
      if (!blockSize || typeof blockSize !== 'number') {
        blockSize = DEFAULT_SIGNAL_BLOCK;
      }
      
      let tobesent = [];
      for (let i = 0; i < blockSize; i++) {
        if (signalBuffer.length > i) {
          tobesent.push(signalBuffer.shift());
        } else {
          break;
        }
      }
      
      if (tobesent.length > 0 && record) {
        wss.of('/client').emit('append signals', {
          signals: tobesent
        });
      }
    })
  } else if (ns === '/app') {
    console.log(`application [${socket.clientId}] connected`);

    /*
     *  Applications
     */
    socket.on('new model', (data) => {
      let processId = getProcessId(data);
      wss.of('/client').emit('new model', {
        processId: processId
      });
    });

    socket.on('append elements', (data) => {
      if (typeof data === 'string') {
        try {
          data = JSON.parse(data)
        } catch (err) {
          console.warn('error parsing received "append elements" data');
          return;
        }
      }

      for (let elem of data.elements) {
        if (!elem.data) elem.data = {};
        elementBuffer.push(elem);
      }
    });

    socket.on('append signals', (data) => {
      if (typeof data === 'string') {
        try {
          data = JSON.parse(data)
        } catch (err) {
          console.warn('error parsing received "append elements" data')
          return;
        }
      }

      var signals = data.signals.filter(n => n != undefined && n != null);

      if (record){
        for (let i = 0; i < signals.length; i++) {
          if (!signals[i]) {
            continue;
          }
          signals[i].time = count;
          count++;

          signalBuffer.push(signals[i]);
        }
      }
    });
  }
});


server.listen(port, function () {
  console.log('collar dev server listening on port', port);
});