first commit

Author: pgte

.eslintignore

@@ -0,0 +1,2 @@
+node_modules
+coverage

.eslintrc.js

@@ -0,0 +1,6 @@
+module.exports = {
+    "extends": "standard",
+    "plugins": [
+        "standard"
+    ]
+};

.istanbul.yml

@@ -0,0 +1,8 @@
+instrumentation:
+    excludes: ['test', 'node_modules']
+check:
+    global:
+        lines: 100
+        branches: 100
+        statements: 100
+        functions: 100

package.json

@@ -1,10 +1,11 @@
 {
   "name": "skiff",
   "version": "1.0.0",
-  "description": "",
-  "main": "index.js",
+  "description": "Raft for Node.js",
+  "main": "src/index.js",
   "scripts": {
-    "test": "echo \"Error: no test specified\" && exit 1"
+    "test": "node --harmony node_modules/istanbul/lib/cli.js cover -- lab -vl && istanbul check-coverage",
+    "style": "eslint src"
   },
   "repository": {
     "type": "git",
@@ -20,5 +21,23 @@
   "bugs": {
     "url": "https://github.com/pgte/skiff/issues"
   },
-  "homepage": "https://github.com/pgte/skiff#readme"
+  "homepage": "https://github.com/pgte/skiff#readme",
+  "devDependencies": {
+    "eslint": "^3.1.1",
+    "eslint-config-standard": "^5.3.5",
+    "eslint-plugin-promise": "^2.0.0",
+    "eslint-plugin-standard": "^2.0.0",
+    "istanbul": "^0.4.4",
+    "pre-commit": "^1.1.3"
+  },
+  "pre-commit": [
+    "style",
+    "test"
+  ],
+  "dependencies": {
+    "debug": "^2.2.0",
+    "msgpack5": "^3.4.0",
+    "multiaddr": "^2.0.2",
+    "reconnect-core": "^1.3.0"
+  }
 }

src/network/index.js

@@ -0,0 +1,7 @@
+const Network = require('./network')
+
+module.exports = createNetwork
+
+function createNetwork (options) {
+  return new Network(options)
+}

src/network/network.js

@@ -0,0 +1,53 @@
+'use strict'
+
+const debug = require('debug')('skiff.network')
+const Duplex = require('stream').Duplex
+const Peer = require('./peer')
+
+const defaultOptions = {
+  objectMode: true,
+  highWaterMark: 50
+}
+
+class Network extends Duplex {
+
+  constructor (_options) {
+    const options = Object.assign({}, _options || {}, defaultOptions)
+    debug('creating network with options %j', options)
+    super(options)
+    this._peers = {}
+    this._options = options
+
+    this.once('finish', this._finish.bind(this))
+  }
+
+  _read (size) {
+    // do nothing
+  }
+
+  _write (message, encoding, callback) {
+    const peer = this._ensurePeer(message.to)
+    peer.write(message, callback)
+  }
+
+  _ensurePeer (address) {
+    debug('ensuring peer %s', address)
+    let peer = this._peers[address]
+    if (!peer) {
+      peer = this._peers[address] = new Peer(address, this._options)
+      peer.on('error', (err) => this.emit('error', err))
+    }
+
+    peer.pipe(this, { end: false }).pipe(peer, { end: false })
+    return peer
+  }
+
+  _finish () {
+    debug('finishing')
+    Object.keys(this._peers).forEach((addr) => this._peers[addr].end())
+    this._peers = {}
+  }
+
+}
+
+module.exports = Network

src/network/peer.js

@@ -0,0 +1,105 @@
+'use strict'
+
+const debug = require('debug')('skiff.peer')
+const Duplex = require('stream').Duplex
+const Multiaddr = require('multiaddr')
+const Msgpack = require('msgpack5')
+
+const reconnect = require('./reconnect')
+
+const defaultOptions = {
+  objectMode: true,
+  highWaterMark: 50
+}
+
+const interestingEvents = [
+  'connect',
+  'reconnect',
+  'disconnect',
+  'error'
+]
+
+const reconnectOptions = {
+  immediate: true
+}
+
+const OK_ERRORS = [
+  'ECONNREFUSED'
+]
+
+class Peer extends Duplex {
+
+  constructor (address, options) {
+    debug('constructing peer from address %j', address)
+    super(Object.assign({}, options || {}, defaultOptions))
+    this._address = Multiaddr(address)
+
+    this.once('finish', this._finish.bind(this))
+    this._connect()
+  }
+
+  _connect () {
+    debug('connecting to %s', this._address)
+
+    this._reconnect = reconnect(reconnectOptions, (peer) => {
+      debug('got stream to peer %s', this._address)
+      const msgpack = Msgpack()
+
+      const toPeer = msgpack.encoder()
+      toPeer.pipe(peer)
+
+      const fromPeer = this._out = msgpack.decoder()
+      fromPeer.pipe(this, { end: false })
+      fromPeer.on('data', (data) => this.push(data))
+
+      peer.on('error', (err) => {
+        if (OK_ERRORS.indexOf(err.code) === -1) {
+          this.emit('error', err)
+        }
+      })
+    })
+    .connect(this._address)
+
+    interestingEvents.forEach((event) => {
+      this._reconnect.on(event, (payload) => {
+        this.emit(event, payload)
+      })
+    })
+
+    this._reconnect
+      .on('connect', (con) => {
+        debug('connected to %s', this._address)
+        con.on('error', (err) => {
+          debug('error from peer:\n%s', err.stack)
+          this.emit('warning', err)
+        })
+      })
+      .on('disconnect', () => {
+        debug('disconnected from %s', this._address)
+        this._out = undefined
+      })
+  }
+
+  _read (size) {
+    // do nothing, we'll emit data when the peer emits data
+  }
+
+  _write (message, encoding, callback) {
+    debug('writing %j to %s', message, this._address)
+    if (this._out) {
+      return this._out.write(message, callback)
+    } else {
+      debug('not connected yet to peer %s', this._address)
+      // if we're not connected we discard the message
+      callback()
+    }
+  }
+
+  _finish () {
+    debug('finishing connection to peer %s', this._address)
+    this._reconnect.disconnect()
+  }
+
+}
+
+module.exports = Peer

src/network/reconnect.js

@@ -0,0 +1,15 @@
+'use strict'
+
+const debug = require('debug')('skiff.reconnect')
+const Reconnect = require('reconnect-core')
+const net = require('net')
+
+module.exports = Reconnect((maddr) => {
+  const nodeAddr = maddr.nodeAddress()
+  const addr = {
+    port: nodeAddr.port,
+    host: nodeAddr.address
+  }
+  debug('connecting to %j', addr)
+  return net.connect(addr)
+})

test/network.js

@@ -0,0 +1,100 @@
+'use strict'
+
+const Lab = require('lab')
+const lab = exports.lab = Lab.script()
+const describe = lab.experiment
+const before = lab.before
+const after = lab.after
+const it = lab.it
+const Code = require('code')
+const expect = Code.expect
+
+const Multiaddr = require('multiaddr')
+const net = require('net')
+const timers = require('timers')
+const Msgpack = require('msgpack5')
+
+const Network = require('../src/network')
+
+const serverAddresses = [
+  '/ip4/127.0.0.1/tcp/8080',
+  '/ip4/127.0.0.1/tcp/8081',
+  '/ip4/127.0.0.1/tcp/8082',
+  ]
+
+describe('network', () => {
+
+  let network, servers
+  let serverData = serverAddresses.map(() => [])
+
+  before(done => {
+    let listening = 0
+
+    servers = serverAddresses.map((addr, index) => {
+      const maddr = Multiaddr(addr)
+      const server = net.createServer(onServerConnection)
+      const listenAddr = maddr.nodeAddress()
+      server.listen({port: listenAddr.port, host: listenAddr.address}, onceListening)
+      return server
+
+      function onServerConnection(conn) {
+        const msgpack = Msgpack()
+        conn.pipe(msgpack.decoder()).on('data', onServerData)
+      }
+
+      function onServerData(data) {
+        serverData[index].push(data)
+      }
+    })
+
+
+    function onceListening() {
+      if (++ listening === servers.length) {
+        done()
+      }
+    }
+  })
+
+  after(done => {
+    let closed = 0
+    servers.forEach(server => server.close(onceClosed))
+    network.end()
+
+    function onceClosed() {
+      if (++ closed === servers.length) {
+        done()
+      }
+    }
+  })
+
+  it('can be instantiated', done => {
+    network = Network()
+    done()
+  })
+
+  it('can be used to send a message to a peer', done => {
+    network.write({to: serverAddresses[0], what: 'hey'}, done)
+  })
+
+  it('peer gets the message', done => {
+    expect(serverData[0]).to.equal([{to: serverAddresses[0], what: 'hey'}])
+    done();
+  })
+
+  it('allows message to unconnected peer', done => {
+    network.write({to: '/ip4/127.0.0.1/tcp/8083', what: 'hey'}, done)
+  })
+
+  it('waits a bit', done => {
+    timers.setTimeout(done, 1000)
+  })
+
+  // it('allows peer to disconnect', done => {
+
+  // })
+
+  // it('allows peer to reconnect', done => {
+
+  // })
+
+})