peer-bridge

.gitignore

@@ -1,2 +1,3 @@
+*.log
 **/node_modules
 src/public/js/bundle.js

gulpfile.js

@@ -0,0 +1,16 @@
+'use strict'
+
+var gulp = require('gulp')
+
+var sigServer = require('./src/sig-server')
+var sigS
+
+gulp.task('test:browser:before', (done) => {
+  sigS = sigServer.start(15555, done)
+})
+
+gulp.task('test:browser:after', (done) => {
+  sigS.stop(done)
+})
+
+require('aegir/gulp')(gulp)

package.json

@@ -4,16 +4,17 @@
   "description": "Real-time music collaboration for the web",
   "main": "src/index.js",
   "scripts": {
-    "build": "browserify -t brfs src/public/js/app.js > src/public/js/bundle.js",
-    "start": "npm run build && node src/index.js",
+    "build": "browserify -t brfs src/app/index.js > src/public/js/bundle.js",
+    "start": "npm run build && node src/sig-server/index.js",
     "lint": "standard",
-    "test": "exit 0"
+    "test": "gulp test",
+    "test:node": "gulp test:node",
+    "test:browser": "gulp test:browser"
   },
   "repository": {
     "type": "git",
     "url": "https://github.com/websound/RemotePerformer.git"
   },
-  "author": "Nathan Ward",
   "license": "MIT",
   "bugs": {
     "url": "https://github.com/websound/RemotePerformer/issues"
@@ -32,14 +33,23 @@
     "inert": "^4.0.0",
     "insert-css": "^0.2.0",
     "jquery": "^2.1.0",
+    "length-prefixed-stream": "^1.5.0",
     "process-nextick-args": "^1.0.7",
+    "simple-peer": "^6.0.4",
+    "socket.io": "^1.4.6",
+    "socket.io-client": "^1.4.6",
     "watchify": "^3.7.0",
     "ws": "^1.1.0"
   },
   "devDependencies": {
+    "aegir": "^3.2.0",
+    "chai": "^3.5.0",
     "eslint-config-standard": "^5.1.0",
     "eslint-plugin-standard": "^1.3.2",
+    "mocha": "^2.5.3",
     "pre-commit": "^1.1.3",
-    "standard": "^7.1.2"
+    "run-parallel": "^1.1.6",
+    "standard": "^7.1.2",
+    "webrtcsupport": "^2.2.0"
   }
 }

src/app/peer-bridge.js

@@ -0,0 +1,117 @@
+var SimplePeer = require('simple-peer')
+var EE = require('events').EventEmitter
+var util = require('util')
+var lpstream = require('length-prefixed-stream')
+var io = require('socket.io-client')
+
+exports = module.exports = PeerBridge
+
+util.inherits(PeerBridge, EE)
+
+function PeerBridge () {
+  if (!(this instanceof PeerBridge)) {
+    return new PeerBridge()
+  }
+
+  this.conns = {}
+  this.id = (~~(Math.random() * 1e9)).toString(36) + Date.now()
+
+  this.setUp = (sigUrl, callback) => {
+    var sioClient = io.connect(sigUrl, {
+      transports: ['websocket'],
+      'force new connection': true
+    })
+    sioClient.once('connect_error', callback)
+    sioClient.on('connect', () => {
+      sioClient.emit('ss-join', this.id)
+      sioClient.on('ws-handshake', incommingDial.bind(this))
+      sioClient.on('ws-peer', peerDiscovered.bind(this))
+      callback()
+    })
+
+    function incommingDial (offer) {
+      if (offer.answer) {
+        return
+      }
+
+      const channel = new SimplePeer({ trickle: false })
+
+      channel.on('connect', () => {
+        this.conns[offer.srcId] = {
+          channel: channel,
+          lps: {
+            encode: lpstream.encode(),
+            decode: lpstream.decode()
+          }
+        }
+
+        this.conns[offer.srcId].lps.encode
+          .pipe(this.conns[offer.srcId].channel)
+          .pipe(this.conns[offer.srcId].lps.decode)
+          .on('data', (data) => {
+            this.emit('msg', data)
+          })
+        this.emit('peer', offer.srcId)
+      })
+
+      channel.on('signal', function (signal) {
+        offer.signal = signal
+        offer.answer = true
+        sioClient.emit('ss-handshake', offer)
+      })
+
+      channel.signal(offer.signal)
+    }
+
+    function peerDiscovered (_id) {
+      var intentId = (~~(Math.random() * 1e9)).toString(36) + Date.now()
+      var channel = new SimplePeer({ initiator: true, trickle: false })
+
+      channel.on('signal', (signal) => {
+        sioClient.emit('ss-handshake', {
+          intentId: intentId,
+          srcId: this.id,
+          dstId: _id,
+          signal: signal
+        })
+      })
+
+      sioClient.on('ws-handshake', (offer) => {
+        if (offer.intentId !== intentId || !offer.answer) {
+          return
+        }
+
+        channel.on('connect', () => {
+          this.conns[_id] = {
+            channel: channel,
+            lps: {
+              encode: lpstream.encode(),
+              decode: lpstream.decode()
+            }
+          }
+
+          this.conns[_id].lps.encode
+            .pipe(this.conns[_id].channel)
+            .pipe(this.conns[_id].lps.decode)
+            .on('data', (data) => {
+              this.emit('msg', data)
+            })
+
+          this.emit('peer', _id)
+        })
+        channel.signal(offer.signal)
+      })
+    }
+  }
+
+  this.broadcast = (buf) => {
+    Object.keys(this.conns).forEach((key) => {
+      this.conns[key].lps.encode.write(buf)
+    })
+  }
+
+  this.send = (id, buf) => {
+    this.conns[id].lps.encode.write(buf)
+  }
+}
+

src/index.js

@@ -1,34 +1,5 @@
-var attachWSS = require('./ws-server.js')
-var Hapi = require('hapi')
-var path = require('path')
+'use strict'
 
-var server = new Hapi.Server({})
+exports = module.exports
 
-server.connection({
-  port: Number(process.env.PORT) || 9090
-})
-
-server.register(require('inert'), function (err) {
-  if (err) {
-    throw err
-  }
-
-  server.route({
-    method: 'GET',
-    path: '/{param*}',
-    handler: {
-      directory: {
-        path: path.join(__dirname, '/public')
-      }
-    }
-  })
-
-  server.start(function (err) {
-    if (err) {
-      throw err
-    }
-
-    console.log('Server running at:', server.info.uri)
-    attachWSS(server)
-  })
-})
+exports.PeerBridge = require('./app/peer-bridge.js')

src/sig-server/index.js

@@ -0,0 +1,54 @@
+var attachWSS = require('./ws-server.js')
+var Hapi = require('hapi')
+var path = require('path')
+
+exports = module.exports
+
+exports.start = (port, callback) => {
+  if (typeof port === 'function') {
+    callback = port
+    port = undefined
+  }
+  var options = {
+    connections: {
+      routes: {
+        cors: true
+      }
+    }
+  }
+
+  var httpListener = new Hapi.Server(options)
+
+  httpListener.connection({
+    port: port || Number(process.env.PORT) || 9090
+  })
+
+  httpListener.register(require('inert'), function (err) {
+    if (err) {
+      throw err
+    }
+
+    httpListener.route({
+      method: 'GET',
+      path: '/{param*}',
+      handler: {
+        directory: {
+          path: path.join(__dirname, '../public')
+        }
+      }
+    })
+
+    httpListener.start(function (err) {
+      if (err) {
+        throw err
+      }
+
+      console.log('Server running at:', httpListener.info.uri)
+      httpListener.peers = attachWSS(httpListener).peers
+      callback(null, httpListener.info)
+    })
+  })
+
+  return httpListener
+}
+

src/sig-server/ws-server.js

@@ -0,0 +1,75 @@
+'use strict'
+
+const log = console.log
+const SocketIO = require('socket.io')
+
+module.exports = (http) => {
+  const io = new SocketIO(http.listener)
+
+  io.on('connection', handle)
+
+  const peers = {}
+
+  this.peers = () => {
+    return peers
+  }
+
+  function safeEmit (addr, event, arg) {
+    const peer = peers[addr]
+    if (!peer) {
+      log('trying to emit %s but peer is gone', event)
+      return
+    }
+
+    peer.emit(event, arg)
+  }
+
+  function handle (socket) {
+    socket.on('ss-join', join.bind(socket))
+    socket.on('ss-leave', leave.bind(socket))
+    socket.on('disconnect', disconnect.bind(socket)) // socket.io own event
+    socket.on('ss-handshake', forwardHandshake)
+  }
+
+  // join this signaling server network
+  function join (id) {
+    peers[id] = this // socket
+    Object.keys(peers).forEach((_id) => {
+      if (_id === id) {
+        return
+      }
+      // broadcast the new peer
+      safeEmit(_id, 'ws-peer', id)
+    })
+  }
+
+  function leave (multiaddr) {
+    if (peers[multiaddr]) {
+      delete peers[multiaddr]
+    }
+  }
+
+  function disconnect () {
+    Object.keys(peers).forEach((mh) => {
+      if (peers[mh].id === this.id) {
+        delete peers[mh]
+      }
+    })
+  }
+
+  // forward an WebRTC offer to another peer
+  function forwardHandshake (offer) {
+    if (offer.answer) {
+      safeEmit(offer.srcId, 'ws-handshake', offer)
+    } else {
+      if (peers[offer.dstId]) {
+        safeEmit(offer.dstId, 'ws-handshake', offer)
+      } else {
+        offer.err = 'peer is not available'
+        safeEmit(offer.srcId, 'ws-handshake', offer)
+      }
+    }
+  }
+
+  return this
+}

test/browser.js

@@ -0,0 +1,3 @@
+'use strict'
+
+require('./peer-bridge/test-peer-bridge.js')

test/node.js

@@ -0,0 +1,3 @@
+'use strict'
+
+require('./sig-server/test-sig-server.js')