<|start_filename|>lib/sd-swim.js<|end_filename|>
'use strict'

// SD-SWIM main module. Starts the protocols

const { EventEmitter } = require('events')
const pino = require('pino')
const propagate = require('propagate')
const Net = require('./net')
const Dissemination = require('./dissemination')
const Failure = require('./failure')
const Join = require('./join')
const Metadata = require('./metadata')
const Members = require('./members')
const { states: { STARTED, STOPPED, JOINED } } = require('./states')

class SDSwim extends EventEmitter {
  constructor (opts = {}) {
    super()
    this.opts = opts
    this.opts.sdswim = this

    // port
    if (opts.port === 0 || !opts.port) {
      // default to 0
      // 0 is a valid value (use random port)
      this.port = 0
    }
    this.port = opts.port

    // logger
    this.logger = opts.logger ? opts.logger : pino()
    if (!opts.logger) {
      this.logger.level = opts.logLevel || 'info'
    }
    opts.logger = this.logger

    // algorithm parameters and their defaults
    this.opts.joinTimeout = this.joinTimeout = opts.joinTimeout || 2000
    this.opts.interval = this.interval = opts.interval || 100
    this.opts.pingTimeout = this.pingTimeout = opts.pingTimeout || 20
    this.opts.pingReqTimeout = this.pingReqTimeout = opts.pingReqTimeout || 60
    this.opts.pingReqGroupSize = this.pingReqGroupSize =
      opts.pingReqGroupSize || 3
    this.opts.updatesMaxSize = this.updatesMaxSize = opts.updatesMaxSize || 50
    this.opts.suspectTimeout = this.suspectTimeout = opts.suspectTimeout || 1000
    this.opts.disseminationFactor = this.disseminationFactor =
      opts.disseminationFactor || 15

    // Metadata distribution sub-protocol
    this.metadataDistributionTimeout = opts.metadataDistributionTimeout || 1000

    // states and initial state
    this.state = STOPPED
    this.opts.initialHosts = this.initialHosts = opts.hosts

    // internal modules
    this.opts.dissemination = this.dissemination = new Dissemination(opts) // dissemination module
    this.opts.net = this.net = new Net(opts) // network comm
    this.opts.members = this.members = new Members(opts) // members list
    this.opts.metadata = this.metadata = new Metadata(opts) // metadata distribution sub-protocol

    // protocols modules
    this.opts.join = this.join = new Join(opts) // join protocol
    this.opts.failure = this.failure = new Failure(opts) // failure detector

    // NET messages managed explicitly
    this.net.on('unknown', (sender, msg) =>
      this.logger.debug(`Unknown Message received ${sender}: ${msg}`)
    )

    this.net.on('up', port => {
      this.port = port
      this.emit('up', port)
    })

    // JOIN events
    this.join.on('join-timeout', err => this.emit('error', err))
    this.join.on('joined', () => {
      this.state = JOINED
      this.logger.debug(`The node is now in state ${this.state}`)
      this.emit('joined')
    })

    // Bubble up events from the protocol modules
    propagate(this.failure, this)
    propagate(this.members, this)
    propagate(this.metadata, this)
  }

  // return the ALIVE and SUSPECT nodes
  get memberList () {
    return this.members.list.map(({ node }) => node)
  }

  get me () {
    return { host: this.host || '127.0.0.1', port: this.port }
  }

  whoami () {
    return Object.assign(this.me, { state: this.state })
  }

  start (cb) {
    const callback = cb || (() => {})
    return new Promise((resolve, reject) => {
      this.net.start((err, port) => {
        if (err) {
          this.emit('error', err)
          callback(err)
          return reject(err)
        }
        this.join.start() // start join protocol
        this.failure.start() // start failure detector
        this.metadata.start() // start failure detector
        this.state = STARTED
        callback(null, port)
        return resolve(port)
      })
    })
  }

  stop (cb) {
    const callback = cb || (() => {})
    return new Promise((resolve, reject) => {
      this.join.stop() // stop join protocol
      this.failure.stop() // stop failure detector
      this.metadata.stop() // start failure detector
      this.net.stop(err => {
        if (err) {
          callback(err)
          return reject(err)
        }
        this.state = STOPPED
        callback()
        return resolve()
      })
    })
  }

  // METADATA API.
  // Add all mine metadata. entries must be in the form [{key, value}]
  addAll (entries) {
    this.metadata.my = entries
  }

  // Add a value to my metadata
  add (key, value) {
    this.metadata.add(key, value)
  }

  remove (key) {
    this.metadata.remove(key)
  }

  // Get all the data, including mine.
  get data () {
    return this.metadata.data
  }
}
module.exports = SDSwim


<|start_filename|>lib/dissemination.js<|end_filename|>
'use strict'

const {sortBy, remove, find, values, keys} = require('lodash')
const {nodeStates: {ALIVE, SUSPECT, FAULTY}} = require('../lib/states')
const uuidV4 = require('uuid/v4')

// Manages the updates to be "disseminated".
// This module maintains:
// - A list of current "updates" to be propagated
// - A list of recent "faulties", which is necessary to avoid that case a node is marked
//   as Faulty (so removed) but there are still "Alive" messages around (so is then readded)
// - A list of "expired" updates, i.e. the updates that has already been gossiped up up
//   to the limit. We have to drop this ones to avoid messages loops.

class Dissemination {
  constructor (opts) {
    this.opts = opts
    this.logger = this.opts.logger
    // Specific Update options
    this.updatesMaxSize = opts.updatesMaxSize || 50
    this.disseminationFactor = opts.disseminationFactor || 15
    this.suspectTimeout = opts.suspectTimeout || 1000

    this._updates = []  // current updates
    this._faulties = {} // recent faulties.
    this._expired = {}  // expired updates uuid
    this.getLimit = (factor, size) => Math.ceil(factor * Math.log(size + 1) / Math.log(10))
    this.key = ({host, port}) => `${host}:${port}`

    // Faulties / expired clean up. Parametrizied using `this.suspectTimeout`
    setInterval(() => {
      const now = Date.now()
      const newExpired = {}
      const newFaulties = {}

      for (const key of keys(this._expired)) {
        if (now > (this._expired[key] + this.suspectTimeout * 5)) {
          newExpired[key] = this._expired[key]
        }
      }

      for (const key of keys(this._faulties)) {
        if (now > (this._faulties[key] + this.suspectTimeout * 2)) {
          newFaulties[key] = this._faulties[key]
        }
      }
      this._expired = newExpired
      this._faulties = newFaulties
    }, this.suspectTimeout * 2)
  }

  /**
   * Get the updates to be propagated, up to `this.updatesMaxSize`.
   * The updates are the one less-gossiped, so first all the updates are ordered by
   * `gossiped` (ascending) and the first updatesMaxSize are returned (when returning,
   * `gossiped` is incremented). The updates for which `gossiped` is > currentLimit are removed
   */
  getUpdates () {
    this._updates = sortBy(this._updates, 'gossiped')
    const ret = []

    for (let i = 0; i < this.updatesMaxSize && i < this._updates.length; i++) {
      this._updates[i].gossiped = this._updates[i].gossiped + 1
      const {node, setBy, state, incNumber, uuid} = this._updates[i]
      ret.push({node, setBy, state, incNumber, uuid})
    }
    // current gossip limit for each update. It dependes on the current group size.
    const currentLimit = Math.ceil(this.disseminationFactor * Math.log(this.opts.members.size() + 1) / Math.log(10))
    const expired = remove(this._updates, el => el.gossiped >= currentLimit)
    for (const exp of expired) {
      this._expired[exp.uuid] = Date.now()
    }
    return ret
  }

  /**
   * The update is added only if not present, otherwise it's updated.
   * If so if we have an SUSPECT update (same node, same incNumber) we can
   * change it to FAULT or ALIVE without having two concurrent updates on the same node
   * propagated from here.
   */
  addUpdate (node, setBy, state, incNumber, uuid) {
    if (uuid && this._expired[uuid]) {
      return  // update expired, must skip it
    }

    const update = find(this._updates,
      el => (el.node.host === node.host && el.node.port === node.port &&
        el.state === state && el.incNumber === incNumber))

    if (!update) {
      if (!uuid) {
        this._updates.push({node, setBy, state, incNumber, gossiped: 0, uuid: uuidV4(null, uuid, 0)})
      } else {
        this._updates.push({node, setBy, state, incNumber, gossiped: 0, uuid})
      }
    }

    if (state === FAULTY) {
      const key = this.key(node)
      if (!this._faulties[key]) {
        this._faulties[key] = Date.now()
      }
    }
  }

  // alias of addUpdate
  propagate ({node, setBy, state, incNumber, uuid}) {
    this.addUpdate(node, setBy, state, incNumber, uuid)
  }

  /**
   * It's possible to have two different updates for the same node.
   * (typically an "old" ALIVE and then a FAULTY).
   * If the FAULTY is processed, the node is removed from the member list, but
   * the following ALIVE update will put it back! To avoid this, we must identify
   * between this batch of updates, the ones of the same node and then for each node,
   * removes the ALIVE / SUSPECT updates if a FAULTY is present.
   * Then we have to process the updates in this order: ALIVE - SUSPECT - FAULTY
   */
  processUpdates (updates) {
    const groupedUpdates = updates.reduce((prev, curr) => {
      const key = this.key(curr.node)
      if (!prev[key]) {
        prev[key] = {alive: [], suspect: [], faulty: []}
      }
      if (curr.state === ALIVE) {
        prev[key].alive.push(curr)
      }
      if (curr.state === SUSPECT) {
        prev[key].suspect.push(curr)
      }
      if (curr.state === FAULTY) {
        prev[key].faulty.push(curr)
      }
      return prev
    }, {})
    const alives = []
    const suspects = []
    const faulties = []
    for (const el of values(groupedUpdates)) {
      if (el.faulty.length !== 0) {
        faulties.push(...el.faulty)
      } else {
        suspects.push(...el.suspect)
        alives.push(...el.alive)
      }
    }
    const filteredAndOrdered = alives.concat(suspects, faulties)
    for (const update of filteredAndOrdered) {
      this.processUpdate(update)
    }
  }

  processUpdate (update) {
    switch (update.state) {
      case 0: this._processAlive(update)
        break
      case 1: this._processSuspect(update)
        break
      case 2: this._processFaulty(update)
        break
    }
  }

  // | Condition                                           |      Member List                    |  Updates                   |
  // |-----------------------------------------------------|:-----------------------------------:|---------------------------:|
  // | Node not present                                    |   Member added as `ALIVE`           |     Propagated             |
  // | Node present and `ALIVE`, with incNumber <= i       |   Member updated (setBy, incNumber) |     Propagated             |
  // | Node present and `ALIVE`, with incNumber >  i       |                                     |     Drop                   |
  // | Node present and `SUSPECTED`, with incNumber <= i   |   Member updated as `ALIVE`         |     Propagated             |
  // | Node present and `SUSPECTED`, with incNumber >  i   |                                     |     Drop                   |
  _processAlive (update) {
    const {node, state, setBy, incNumber} = update
    const key = this.key(node)
    const member = this.opts.members.findMember(node)
    // if node not present, is added as ALIVE
    if (!member) {
      if (!this._faulties[key]) {
        this.logger.debug(update, 'adding new ALIVE member')
        this.opts.members.addOrUpdateMember(node, ALIVE, setBy, incNumber)
        return this.propagate(update)
      }
      return
    }
    if (member.incNumber > incNumber) {
      this.logger.debug(update, 'Incarnation number of the member greater than the one from the message, dropped')
      return
    }
    if (member.state === ALIVE) {
      this.logger.debug(update, 'update ALIVE member')
      this.opts.members.addOrUpdateMember(node, state, setBy, incNumber)
    }

    if (member.state === SUSPECT) {
      this.logger.debug(update, 'update ALIVE member from SUSPECT')
      this.opts.members.addOrUpdateMember(node, state, setBy, incNumber)
    }
    this.propagate(update)
  }

  // | Condition                                             |      Member List                    |  Updates                   |
  // |-------------------------------------------------------|:-----------------------------------:|---------------------------:|
  // | Node is me                                            |   incNumber is incremented          |     new `ALIVE` update created   |
  // | Node not present                                      |   Member added as `SUSPECT`         |     Propagated                   |
  // | Member present and `ALIVE`, with incNumber <= i       |   Member changed to `SUSPECT`       |     Propagated                   |
  // | Member present and `ALIVE`, with incNumber  > i       |                                     |     Drop                         |
  // | Member present and `SUSPECTED`, with incNumber <=  i  |   Member updated (setBy, incNumber) |     Propagated                   |
  // | Member present and `SUSPECTED`, with incNumber >  i   |                                     |     Drop                         |
  _processSuspect (update) {
    const {node, state, setBy, incNumber} = update
    const member = this.opts.members.findMember(node)
    const isMe = this.opts.members.isMe(node)
    const key = this.key(node)

    // Node is me
    if (isMe) {
      this.logger.debug(update, 'node is me, incrementing incNumber and send ALIVE update')
      const me = this.opts.members.me()
      const newIncNumber = incNumber + 1
      this.opts.members.addOrUpdateMember(node, ALIVE, me, newIncNumber)
      return this.propagate({node, state: ALIVE, setBy: me, incNumber: newIncNumber})
    }

    // if node not present, is added as SUSPECT
    if (!member) {
      if (!this._faulties[key]) {
        this.logger.debug(update, 'adding new SUSPECT member')
        this.opts.members.addOrUpdateMember(node, SUSPECT, setBy, incNumber)
        return this.propagate(update)
      }
      return
    }
    if (member.incNumber > incNumber) {
      this.logger.debug(update, 'Incarnation number of the member greater than the one from the message, dropped')
      return
    }
    if (member.state === ALIVE) {
      this.logger.debug(update, 'update to SUSPECT')
      this.opts.members.addOrUpdateMember(node, SUSPECT, setBy, incNumber)
    }

    if (member.state === SUSPECT) {
      this.logger.debug(update, 'update SUSPECT member')
      this.opts.members.addOrUpdateMember(node, state, setBy, incNumber)
    }
    this.propagate(update)
  }

  // | Condition                                           |      Member List                    |  Updates                   |
  // |-----------------------------------------------------|:-----------------------------------:|---------------------------:|
  // | Node not present                                    |                                     |     Propagated                    |
  // | Node is me                                          |   incNumber is incremented          |     new `ALIVE` update created    |
  // | Node present                                        |   remove from the alive nodes       |     Propagated                    |
  _processFaulty (update) {
    const {node, incNumber} = update
    const member = this.opts.members.findMember(node)
    const isMe = this.opts.members.isMe(node)

    // Node is me, I'm ALIVE!
    if (isMe) {
      this.logger.debug(update, 'node is me, incrementing incNumber and send ALIVE update')
      const me = this.opts.members.me()
      const newIncNumber = incNumber + 1
      this.opts.members.addOrUpdateMember(node, ALIVE, me, newIncNumber)
      return this.propagate({node, state: ALIVE, setBy: me, incNumber: newIncNumber})
    }
    // if node is present, removes it
    if (member) {
      this.opts.members.removeMember(node)
    }
    this.propagate(update)
  }
}

module.exports = Dissemination


<|start_filename|>lib/net.js<|end_filename|>
'use strict'

// Network communication.
// If present, sens the updates when sending ping, ping-req and ack,
// and process updates if present when receiving messages

const {EventEmitter} = require('events')
const dgram = require('dgram')
const Messages = require('./messages')

class Net extends EventEmitter {
  constructor (opts) {
    super()
    this.opts = opts
    this.port = opts.port
    this.logger = opts.logger
    this.dissemination = this.opts.dissemination
    this.messages = new Messages(opts)
    this.socket = dgram.createSocket('udp4')
  }

  start (cb) {
    if (!this.socket) { // after a close()
      this.socket = dgram.createSocket('udp4')
    }

    this.socket.on('message', (message, remote) => {
      const {address: host, port} = remote
      const msg = this.messages.decodeMessage(message)
      let eventMessage
      switch (msg.type) {
        case this.messages.types.JOIN:
          eventMessage = 'join'
          break
        case this.messages.types.JOIN_ACK:
          eventMessage = 'join-ack'
          break
        case this.messages.types.PING:
          eventMessage = 'ping'
          break
        case this.messages.types.PING_REQ:
          eventMessage = 'ping-req'
          break
        case this.messages.types.ACK:
          eventMessage = 'ack'
          break
        case this.messages.types.META:
          eventMessage = 'meta'
          break
        case this.messages.types.ALL_META:
          eventMessage = 'all-meta'
          break
        default:
          eventMessage = 'unknown'
      }
      this.dissemination.processUpdates(msg.updates)
      return this.emit(eventMessage, {sender: {host, port}}, msg)
    })

    this.socket.on('error', err => {
      if (cb) {
        return cb(err)
      }
      return this.emit('error', err)
    })

    this.socket.on('listening', () => {
      const {port} = this.socket.address()
      if (cb) {
        cb(null, port)
      }
      return this.emit('up', port)
    })

    return this.socket.bind(this.port)
  }

  stop (cb) {
    if (!this.socket) {
      return cb()
    }
    this.socket.close(err => {
      if (err) {
        return cb(err)
      }
      this.socket = null // cannot reuse
      return cb()
    })
  }

  _send (msg) {
    if (!this.socket) { // if closed
      return
    }
    const {host, port, type, data} = msg
    this.logger.trace(`Sending message ${data} to ${host}:${port}`)
    this.socket.send(data, 0, data.length, port, host, err => {
    // We have to catch the condition when the socked is closed "while" sending
      if ((!err || (err.message && err.message.include('ECANCELED'))) && !this.socket) {
        this.logger.debug(`Message of type ${type} sent to ${host}:${port}`)
      }
      // Also  we should manage the condition when host is not ENOTFOUND (the name can be logic)
      if (err && (err.message && err.message.include('ENOTFOUND'))) {
        this.logger.error(err)
      }
      if (err) throw err
    })
  }

  _sendMessages (messages) {
    for (const msg of messages) {
      this._send(msg)
    }
  }

  sendJoin (initialHosts) {
    this._sendMessages(this.messages.joinMessages(initialHosts))
  }

  sendJoinAckMessage (destination, members) {
    this._send(this.messages.joinAckMessage(destination, members))
  }

  sendPing (destination) {
    this._send(this.messages.pingMessage(destination, this.dissemination.getUpdates()))
  }

  sendPingReq (destination, target, requester) {
    this._send(this.messages.pingReqMessage(destination, target, requester, this.dissemination.getUpdates()))
  }

  sendAck (destination, target, requester) {
    this._send(this.messages.ackMessage(destination, target, requester, this.dissemination.getUpdates()))
  }

  sendMeta (nodes, metadata) {
    this._sendMessages(this.messages.metaMessages(nodes, metadata))
  }

  sendAskAllMeta (destination, metadata) {
    this._send(this.messages.askAllMetaMessage(destination))
  }
}

module.exports = Net


<|start_filename|>test/failure.js<|end_filename|>
/* eslint no-console:0 */

const Lab = require('lab')
const lab = exports.lab = Lab.script()
const {describe, it, beforeEach, afterEach} = lab
const SDSwim = require('../lib/sd-swim')
const {states: {JOINED}} = require('../lib/states')
const pino = require('pino')
const assert = require('power-assert')
const {startNodes, stopNodes, delay, compareNodesLists} = require('./common')
const {cloneDeep} = require('lodash')

describe('Failure Detector', () => {
  describe('given a started node', () => {
    let target, node1, node2
    const nodeOpts = [{port: 12340}]

    beforeEach(() => startNodes(nodeOpts).then(results => {
      [target] = results
    }))

    afterEach(() => stopNodes([target, node1, node2]))

    it('should send a ping message to target after join', done => {
      const hosts = [{host: '127.0.0.1', port: target.port}]

      // start a single node that join the target.
      const sdswim = new SDSwim({logger: pino(), port: 12341, hosts})
      sdswim.on('joined', () => {
        const myself = sdswim.whoami()
        assert.strictEqual(myself.state, JOINED)
      })

      sdswim.on('ping', target => {
        assert.strictEqual(target.port, nodeOpts[0].port)
        sdswim.stop(() => {
          done()
        })
      })

      sdswim.on('ack', target => {
        assert.strictEqual(target.port, nodeOpts[0].port)
        sdswim.stop(() => {
          done()
        })
      })

      sdswim.start()
    })

    it('should generate peerUp and peerDown events', done => {
      const hosts = [{host: '127.0.0.1', port: target.port}]

      // start node1. start node2. stop node2.
      // Check that the events are emitted correctly in node1
      node1 = new SDSwim({logger: pino(), port: 12341, hosts})
      node2 = new SDSwim({logger: pino(), port: 12342, hosts})

      node1.on('peerUp', ({port}) => {
        assert.strictEqual(port, node2.port)
      })

      node1.on('peerDown', ({port}) => {
        assert.strictEqual(port, node2.port)
        done()
      })

      node1.start()
      .then(delay(1000))
      .then(() => node2.start())
      .then(delay(1000))
      .then(() => node2.stop())
      .then(() => {
      })
    })
  })

  describe('given 5 started nodes', () => {
    let nodes
    const opts =
      [{port: 12340},
       {port: 12341, hosts: [{host: '127.0.0.1', port: 12340}], suspectTimeout: 500},
       {port: 12342, hosts: [{host: '127.0.0.1', port: 12340}], suspectTimeout: 500},
       {port: 12343, hosts: [{host: '127.0.0.1', port: 12340}], suspectTimeout: 500},
       {port: 12344, hosts: [{host: '127.0.0.1', port: 12340}], suspectTimeout: 500},
       {port: 12345, hosts: [{host: '127.0.0.1', port: 12340}], suspectTimeout: 500}]

    beforeEach(() => startNodes(opts).then(results => {
      nodes = results
    }))

    afterEach(() => stopNodes(nodes))

    it('should start 5 nodes, and then stop them one by one and the member lists must be coherent', () => {
      const expected =
        [ { host: '127.0.0.1', port: 12340 },
        { host: '127.0.0.1', port: 12341 },
        { host: '127.0.0.1', port: 12342 },
        { host: '127.0.0.1', port: 12343 },
        { host: '127.0.0.1', port: 12344 },
        { host: '127.0.0.1', port: 12345 } ]

      const waitStart = delay(1000) // Startup of all nodes
      const waitClose = delay(1500) // must be higher than the 'suspect' timeout

      // check the member list after 2 secs
      return waitStart()
        .then(() => {
          compareNodesLists(nodes[0].memberList, expected)
          compareNodesLists(nodes[1].memberList, expected)
          compareNodesLists(nodes[2].memberList, expected)
          compareNodesLists(nodes[3].memberList, expected)
          compareNodesLists(nodes[4].memberList, expected)
          compareNodesLists(nodes[5].memberList, expected)
          // All memebres lists are OK. now stop and check the member list after every stop
          return nodes[5].stop()
        })
        .then(waitClose)
        .then(() => {
          expected.pop()
          compareNodesLists(nodes[0].memberList, expected)
          compareNodesLists(nodes[1].memberList, expected)
          compareNodesLists(nodes[2].memberList, expected)
          compareNodesLists(nodes[3].memberList, expected)
          compareNodesLists(nodes[4].memberList, expected)
          return nodes[4].stop()
        })
        .then(waitClose)
        .then(() => {
          expected.pop()
          compareNodesLists(nodes[0].memberList, expected)
          compareNodesLists(nodes[1].memberList, expected)
          compareNodesLists(nodes[2].memberList, expected)
          compareNodesLists(nodes[3].memberList, expected)
          return nodes[3].stop()
        })
        .then(waitClose)
        .then(() => {
          expected.pop()
          compareNodesLists(nodes[0].memberList, expected)
          compareNodesLists(nodes[1].memberList, expected)
          compareNodesLists(nodes[2].memberList, expected)
          return nodes[2].stop()
        })
        .then(waitClose)
        .then(() => {
          expected.pop()
          compareNodesLists(nodes[0].memberList, expected)
          compareNodesLists(nodes[1].memberList, expected)
          return nodes[1].stop()
        })
        .then(waitClose)
        .then(() => {
          expected.pop()
          compareNodesLists(nodes[0].memberList, expected)
          return nodes[0].stop()
        })
    })

    it('should start 5 nodes, and then stop and restart two of them and the member lists must be coherent', () => {
      const expected =
        [ { host: '127.0.0.1', port: 12340 },
          { host: '127.0.0.1', port: 12341 },
          { host: '127.0.0.1', port: 12342 },
          { host: '127.0.0.1', port: 12343 },
          { host: '127.0.0.1', port: 12344 },
          { host: '127.0.0.1', port: 12345 } ]

      const afterFirstRestart = [expected[0], expected[1], expected[2], expected[3], expected[5]]
      const afterSecondRestart = cloneDeep(expected)

      const waitStart = delay(1000) // Startup of all nodes
      const wait = delay(2200) // must be higher than the 'fault node' timeout (2x suspect timeout)

      let new4, new5

      // check the member list after 2 secs
      return waitStart()
        .then(() => {
          compareNodesLists(nodes[0].memberList, expected)
          compareNodesLists(nodes[1].memberList, expected)
          compareNodesLists(nodes[2].memberList, expected)
          compareNodesLists(nodes[3].memberList, expected)
          compareNodesLists(nodes[4].memberList, expected)
          compareNodesLists(nodes[5].memberList, expected)
          return nodes[5].stop()
        })
        .then(wait)
        .then(() => {
          expected.pop()
          compareNodesLists(nodes[0].memberList, expected)
          compareNodesLists(nodes[1].memberList, expected)
          compareNodesLists(nodes[2].memberList, expected)
          compareNodesLists(nodes[3].memberList, expected)
          compareNodesLists(nodes[4].memberList, expected)
          return nodes[4].stop()
        })
        .then(wait)
        .then(() => {
          expected.pop()
          compareNodesLists(nodes[0].memberList, expected)
          compareNodesLists(nodes[1].memberList, expected)
          compareNodesLists(nodes[2].memberList, expected)
          compareNodesLists(nodes[3].memberList, expected)
          new5 = new SDSwim(opts[5]) // If I restart nodes[5] I have issues with the previous state: TODO: in close clean all
          nodes.push(new5)
          return new5.start()
        })
        .then(wait)
        .then(() => {
          compareNodesLists(nodes[0].memberList, afterFirstRestart)
          compareNodesLists(nodes[1].memberList, afterFirstRestart)
          compareNodesLists(nodes[2].memberList, afterFirstRestart)
          compareNodesLists(nodes[3].memberList, afterFirstRestart)
          compareNodesLists(new5.memberList, afterFirstRestart)
          new4 = new SDSwim(opts[4]) // If I restart nodes[5] I have issues with the previous state: TODO: in close clean all
          nodes.push(new4)
          return new4.start()
        })
        .then(wait)
        .then(() => {
          compareNodesLists(nodes[0].memberList, afterSecondRestart)
          compareNodesLists(nodes[1].memberList, afterSecondRestart)
          compareNodesLists(nodes[2].memberList, afterSecondRestart)
          compareNodesLists(nodes[3].memberList, afterSecondRestart)
          compareNodesLists(new4.memberList, afterSecondRestart)
          compareNodesLists(new5.memberList, afterSecondRestart)
        })
    })
  })
})


<|start_filename|>lib/messages.js<|end_filename|>
'use strict'

// Encodes and Decodes the protocol buffer messages
const path = require('path')
const { readFileSync } = require('fs')
const protobuf = require('protocol-buffers')
const { Msg: { decode, encode }, MessageType } = protobuf(
  readFileSync(path.join(__dirname, 'proto/sd-swim.proto'))
)

class Messages {
  constructor (opts) {
    this.opts = opts
    this.types = MessageType
  }

  joinMessages (destinations) {
    return destinations.map(({ host, port }) => {
      const type = MessageType.JOIN
      const data = encode({
        destination: { host, port },
        type
      })
      return { host, port, type, data }
    })
  }

  joinAckMessage (destination, initialMemberList) {
    const { host, port } = destination
    const data = encode({
      destination,
      members: initialMemberList,
      type: MessageType.JOIN_ACK
    })
    return { host, port, type: MessageType.JOIN_ACK, data }
  }

  pingMessage (destination, updates) {
    const { host, port } = destination
    const data = encode({ destination, type: MessageType.PING, updates })
    return { host, port, type: MessageType.PING, data }
  }

  pingReqMessage (destination, target, requester, updates) {
    const { host, port } = destination
    const request = { target, requester }
    const data = encode({
      destination,
      type: MessageType.PING_REQ,
      request,
      updates
    })
    return { host, port, type: MessageType.PING_REQ, data }
  }

  ackMessage (destination, target, requester, updates) {
    const { host, port } = destination
    let request
    if (target) {
      // ping ack do not have target/requester
      request = { target, requester }
    }
    const data = encode({
      destination,
      type: MessageType.ACK,
      request,
      updates
    })
    return { host, port, type: MessageType.ACK, data }
  }

  metaMessages (destinations, metadata) {
    return destinations.map(({ host, port }) => {
      const type = MessageType.META
      const data = encode({
        type,
        metadata
      })
      return { host, port, type, data }
    })
  }

  askAllMetaMessage (destination) {
    const { host, port } = destination
    const data = encode({ destination, type: MessageType.ALL_META })
    return { host, port, type: MessageType.ALL_META, data }
  }

  decodeMessage (message) {
    return decode(message)
  }
}

module.exports = Messages


<|start_filename|>lib/failure.js<|end_filename|>
'use strict'

// Failure Detector Module.
// Events emitted:
// - ping:           when ping is sent.                                 Arg: target
// - ping-req:       when ping is triggered (multiple messages sent).   Arg: target
// - ack:            when ack is received.                              Arg: sender

const {EventEmitter} = require('events')
const {values} = require('lodash')

class Failure extends EventEmitter {
  constructor (opts) {
    super()
    this.members = opts.members
    this.logger = opts.logger
    this.net = opts.net

    // Failure detector options
    this.interval = opts.interval || 100
    this.pingTimeout = opts.pingTimeout || 20
    this.pingReqTimeout = opts.pingReqTimeout || 60
    this.pingReqGroupSize = opts.pingReqGroupSize || 3

    // NET messages managed by this module
    this.net.on('ping', (sender, msg) => this.pingReceived(sender, msg))
    this.net.on('ping-req', (sender, msg) => this.pingReqReceived(sender, msg))
    this.net.on('ack', (sender, msg) => this.ackReceived(sender, msg))

    // Active timers
    this.pingTimers = {}
    this.pingReqTimers = {}
  }

  /**
   * Starts the failure detection algorithm
   */
  start () {
    this.pingInterval = setInterval(this._ping.bind(this), this.interval)
  }

  stop () {
    clearInterval(this.pingInterval)
    values(this.pingTimers).forEach(timer => clearInterval(timer))
    values(this.pingReqTimers).forEach(timer => clearInterval(timer))
  }

  _key ({host, port}) {
    return `${host}:${port}`
  }

  /**
   * - Check if member list is empty. If so, exit
   */
  _ping () {
    const member = this.members.getPingableMember()
    if (!member) { return } // no candidates
    const targetNode = member.node
    this.logger.debug(targetNode, 'Member selected for ping')
    const key = this._key(targetNode)

    this.pingTimers[key] = setTimeout(() => {
      delete this.pingTimers[key]            // remove ping timer
      this._pingReq(targetNode)              // trigger ping-req
    }, this.pingTimeout)

    this.net.sendPing(targetNode)
    this.emit('ping', targetNode)
  }

 /**
  * Trigger the ping-req and set a timeout. If this timeout expires, the node is
  * then
  */
  _pingReq (target) {
    this.logger.debug(target, `ping-req`)
    // Get pingable members, escluding the the target
    const pingableMembers = this.members.getPingReqGroupMembers(target, this.pingReqGroupSize)

    // NOTE: the ping req protocol make sense only if pingableMembers.length !== 0
    // If 0 (this WILL happen when two nodes), we will worngly mark directy as SUSPECT
    if (pingableMembers.length === 0) {
      this.logger.debug(target, `Pingable members list empty, so marking the node as SUSPECT directly without any timer`)
      return this.members.updateToSuspect(target)
    }

     // We set only one timeout for all the requests
    const key = this._key(target)
    this.pingReqTimers[key] = setTimeout(() => {
      delete this.pingReqTimers[key]
      this.members.updateToSuspect(target) // NO ACK RECEIVED YET => UPDATE TO SUSPECT
    }, this.pingReqTimeout)
    for (const {node} of pingableMembers) {
      this.net.sendPingReq(node, target, this.members.me)
    }
    this.emit('ping-req', target)
  }

  // Receive a PING message, simply answer with ACK
  // `msg.request` is populated if it a ping subsequent to a ping-req
  pingReceived ({sender}, msg) {
    this.logger.debug(`Received PING message from ${sender.host}:${sender.port} of type: ${msg.type}, answering ACK`)
    if (!msg.request) {
      return this.net.sendAck(sender)
    }
    this.net.sendAck(sender, msg.request.target, msg.request.requester)
  }

  // Receive a PING-REQ message. Send a PING to TARGET.
  // In this case, no timeout is needed. If an answer is received is then propagated
  // back to the requester
  pingReqReceived ({sender: {host, port}}, msg) {
    this.logger.debug(msg, `Received PING-REQ message from ${host}:${port} of type: ${msg.type}`)
    const target = msg.request.target
    const requester = msg.request.requester
    this.net.sendPing(target, target, requester) // in this case destination === target. This can be optimized
  }

  /**
   * When an ack is received, if it's for a PING, the related timer must be deleted,
   * abd if the member is SUSPECT, it mus be changed to ALIVE.
   * If the ack is from PING-REQ, two cases:
   * - I'm the requester of the ping-req, so the final destination of the ack =>
   *     - clear the interval for the ping-req
   *     - if suspect, update to alive
   * - I'm not the requester of the ping-req, the ack is rerouted to the original requester
   */
  ackReceived ({sender}, msg) {
    this.logger.debug(msg, `Received ACK message from ${sender.host}:${sender.port} of type: ${msg.type}`)
    this.emit('ack', sender)
    const key = this._key(sender)

    // ACK from PING
    if (this.pingTimers[key]) {
      clearTimeout(this.pingTimers[key])
      delete this.pingTimers[key]
      this.members.updateToAlive(sender)
      return
    }

    // ACK from PING-REQ
    if (msg.request) {
      if (this.members.isMe(msg.request.requester)) {
        // I'm the original requester
        const key = this._key(msg.request.target)
        if (this.pingReqTimers[key]) {
          clearTimeout(this.pingReqTimers[key])
          delete this.pingReqTimers[key]
          this.members.updateToAlive(sender)
          return
        }
      }
      // If target !== me => we reroute the ack to the original requester
      this.net.sendAck(msg.request.requester, msg.request.target, msg.request.requester)
    }
  }
}
module.exports = Failure


<|start_filename|>test/sd-swim.js<|end_filename|>
/* eslint no-console:0 */

const assert = require('power-assert')
const pino = require('pino')
const Lab = require('lab')
const lab = exports.lab = Lab.script()

const {
  describe,
  it
} = lab
const SDSwim = require('../lib/sd-swim')
const {
  states: {
    STARTED,
    STOPPED
  }
} = require('../lib/states')

describe('SD-Swim', () => {
  it('should start a sd-swim node using default port (0)', done => {
    // start a single node, that should know only his port.
    // Default port:
    const sdswim = new SDSwim({
      logger: pino(),
      port: 0
    })
    sdswim.on('up', port => {
      const myself = sdswim.whoami()
      assert.strictEqual(myself.host, '127.0.0.1')
      assert.strictEqual(myself.port, port)
      assert.strictEqual(myself.state, STARTED)
      sdswim.stop(() => {
        assert.strictEqual(sdswim.whoami().state, STOPPED)
        done()
      })
    })
    sdswim.start()
  })

  it('should start a sd-swim node using random port (0)', done => {
    // start a single node, that should know only his port.
    // Default port:
    const sdswim = new SDSwim({
      logger: pino(),
      port: 0
    })
    sdswim.on('up', port => {
      const myself = sdswim.whoami()
      assert.strictEqual(myself.host, '127.0.0.1')
      assert.strictEqual(myself.port, port)
      assert.strictEqual(myself.state, STARTED)
      sdswim.stop(() => {
        assert.strictEqual(sdswim.whoami().state, STOPPED)
        done()
      })
    })
    sdswim.start()
  })

  it('should start a sd-swim node passing a port', done => {
    // start a single node, that should know only his port.
    const port = 12345
    const sdswim = new SDSwim({
      port
    })
    sdswim.on('up', () => {
      const myself = sdswim.whoami()
      assert.strictEqual(myself.host, '127.0.0.1')
      assert.strictEqual(myself.port, port)
      assert.strictEqual(sdswim.whoami().state, STARTED)
      const me = sdswim.me
      assert.strictEqual(me.host, '127.0.0.1')
      assert.strictEqual(me.port, port)
      sdswim.stop(done)
    })
    sdswim.start()
  })

  it('should start a sd-swim node using a callback', done => {
    const port = 12345
    const sdswim = new SDSwim({
      port
    })
    sdswim.start((err, port) => {
      if (err) {
        return assert.fail(err)
      }
      const myself = sdswim.whoami()
      assert.strictEqual(myself.host, '127.0.0.1')
      assert.strictEqual(myself.port, port)
      sdswim.stop(done)
    })
  })

  it('should fail starting on the same port, checking event', done => {
    const port = 12345
    const sdswim = new SDSwim({
      port
    })
    sdswim.start(() => {
      const sdswim2 = new SDSwim({
        port
      }) // same port
      sdswim2.start()
      sdswim2.on('error', err => {
        assert.strictEqual(err.code, 'EADDRINUSE')
        assert.strictEqual(sdswim2.whoami().state, STOPPED)
        sdswim.stop(done)
      })
    })
  })

  it('should fail starting on the same port, using callbacks', done => {
    const port = 12345
    const sdswim = new SDSwim({
      port
    })
    sdswim.start(() => {
      const sdswim2 = new SDSwim({
        port
      }) // same port
      sdswim2.on('error', () => {})
      sdswim2.start(err => {
        assert.strictEqual(err.code, 'EADDRINUSE')
        assert.strictEqual(sdswim2.whoami().state, STOPPED)
        sdswim.stop(done)
      })
    })
  })

  it('should fail start, stop and then start again correctly', done => {
    const port = 12345
    const sdswim = new SDSwim({
      port
    })
    sdswim.start((err, port) => {
      assert.equal(err, null)
      const myself = sdswim.whoami()
      assert.strictEqual(myself.host, '127.0.0.1')
      assert.strictEqual(myself.port, port)
      assert.strictEqual(myself.state, STARTED)
      sdswim.stop(() => {
        assert.strictEqual(sdswim.whoami().state, STOPPED)
        sdswim.start(() => {
          assert.strictEqual(sdswim.whoami().state, STARTED)
          sdswim.stop(() => {
            assert.strictEqual(sdswim.whoami().state, STOPPED)
            done()
          })
        })
      })
    })
  })

  it('should start a sd-swim node using promises', () => {
    const sdswim = new SDSwim({
      logger: pino()
    })
    return sdswim.start().then(() => sdswim.stop())
  })

  it('should fail starting on the same port, using promises', done => {
    const port = 12345
    const sdswim = new SDSwim({
      port
    })
    const sdswim2 = new SDSwim({
      port
    }) // same port
    sdswim2.on('error', () => {})
    sdswim.start()
      .then(() => sdswim2.start())
      .then(() => assert.fail('Must generate an error (two nodeswith the same port)'))
      .catch(err => {
        assert.strictEqual(err.code, 'EADDRINUSE')
        assert.strictEqual(sdswim2.whoami().state, STOPPED)
        return sdswim.stop(done)
      })
  })

  it('should stop a sd-swim node twice without errors', () => {
    const sdswim = new SDSwim({
      logger: pino()
    })
    return sdswim.start().then(() => sdswim.stop()).then(() => sdswim.stop())
  })

  it('should connect two nodes using defaults (127.0.0.1 and port 0)', done => {
    let a, b
    a = new SDSwim()
    a.on('up', function () {
      b = new SDSwim({hosts: [a.me]})
      b.start()
      b.on('updated-members', membersList => {
        assert(membersList.length === 2)
        a.stop(() => {
          b.stop(done)
        })
      })
    })
    a.start()
  })
})


<|start_filename|>test/messages.js<|end_filename|>
/* eslint no-console:0 */

const Lab = require('lab')
const lab = exports.lab = Lab.script()
const assert = require('power-assert')
const Messages = require('../lib/messages')
const uuidV4 = require('uuid/v4')
const {describe, it} = lab

describe('Messages', () => {
  const host1 = {host: 'host1', port: 1234}
  const host2 = {host: 'host2', port: 5678}
  const host3 = {host: 'host3', port: 9101}
  const members = []
  const uuid1 = Buffer.alloc(16)
  const uuid2 = Buffer.alloc(16)
  uuidV4(null, uuid1, 0)
  uuidV4(null, uuid2, 0)
  members.push({node: host1, state: 0, setBy: host3, incNumber: 0, uuid: uuid1})
  members.push({node: host2, state: 0, setBy: host3, incNumber: 0, uuid: uuid2})
  const updates = members

  const messages = new Messages()

  it('should create a Join message correctly', done => {
    const joinMessages = messages.joinMessages([host1, host2])
    assert.equal(joinMessages.length, 2)
    const message1 = messages.decodeMessage(joinMessages[0].data)
    const message2 = messages.decodeMessage(joinMessages[1].data)
    assert.deepEqual(message1.destination, host1)
    assert.deepEqual(message1.members, [])
    assert.deepEqual(message2.destination, host2)
    assert.deepEqual(message2.members, [])
    done()
  })

  it('should create an JoinAck message correctly', done => {
    const joinAckMessage = messages.joinAckMessage(host1, members)
    const message = messages.decodeMessage(joinAckMessage.data)
    assert.equal(message.type, 1)
    assert.deepEqual(message.destination, host1)
    assert.deepEqual(message.members, members)
    done()
  })

  it('should create a Ping message correctly', done => {
    const pingMessage = messages.pingMessage(host1, updates)
    const message = messages.decodeMessage(pingMessage.data)
    assert.equal(message.type, 2)
    assert.deepEqual(message.destination, host1)
    assert.deepEqual(message.updates, updates)
    done()
  })

  it('should create an Ack message correctly', done => {
    const pingMessage = messages.ackMessage(host1, null, null, updates)
    const message = messages.decodeMessage(pingMessage.data)
    assert.equal(message.type, 3)
    assert.deepEqual(message.destination, host1)
    assert.deepEqual(message.updates, updates)
    done()
  })

  it('should create a PingReq message correctly', done => {
    const pingMessage = messages.pingReqMessage(host1, host2, host3, updates)
    const message = messages.decodeMessage(pingMessage.data)
    assert.equal(message.type, 4)
    assert.deepEqual(message.destination, host1)
    assert.deepEqual(message.updates, updates)
    assert.deepEqual(message.request.target, host2)
    assert.deepEqual(message.request.requester, host3)
    done()
  })

  it('should create an Ack from a PingReq message correctly', done => {
    const pingMessage = messages.ackMessage(host1, host2, host3, updates)
    const message = messages.decodeMessage(pingMessage.data)
    assert.equal(message.type, 3)
    assert.deepEqual(message.destination, host1)
    assert.deepEqual(message.updates, updates)
    assert.deepEqual(message.request.target, host2)
    assert.deepEqual(message.request.requester, host3)
    done()
  })

  // Metadata extension messages ********************+

  it('should create an Meta messages correctly', done => {
    const metadata = [{
      owner: host1,
      entries: [
        {key: 'test1', value: Buffer.from('testValue1')},
        {key: 'test2', value: Buffer.from('testValue2')}
      ],
      version: 3
    }]
    const metaMessages = messages.metaMessages([host1], metadata)
    const message = messages.decodeMessage(metaMessages[0].data)
    assert.equal(message.type, 10)
    assert.deepEqual(message.metadata, metadata)
    done()
  })

  it('should create an ask All Meta message correctly', done => {
    const metaMessage = messages.askAllMetaMessage(host1)
    const message = messages.decodeMessage(metaMessage.data)
    assert.equal(message.type, 11)
    assert.deepEqual(message.destination, host1)
    done()
  })
})


<|start_filename|>package.json<|end_filename|>
{
  "name": "sd-swim",
  "version": "0.6.0",
  "description": "Self- discovery version of SWIM membership protocol",
  "main": "index.js",
  "scripts": {
    "start": "node index.js",
    "test": "node --no-warnings ./node_modules/.bin/lab",
    "ci": "lab -r console -o stdout -r lcov -o ./coverage/lcov.info && cat ./coverage/lcov.info | ./node_modules/.bin/coveralls ./coverage/lcov.info",
    "lint": "eslint .",
    "depcheck": "dependency-check package.json --unused --no-dev --entry test/*.js --entry index.js",
    "nsp": "nsp check",
    "prepush": "npm run nsp && npm run depcheck && npm run lint && npm test"
  },
  "repository": {
    "type": "git",
    "url": "git+https://github.com/marcopiraccini/sd-swim.git"
  },
  "keywords": [
    "SWIM",
    "membership",
    "discovery"
  ],
  "author": {
    "name": "<NAME>",
    "email": "<EMAIL>"
  },
  "license": "MIT",
  "bugs": {
    "url": "https://github.com/marcopiraccini/sd-swim/issues"
  },
  "homepage": "https://github.com/marcopiraccini/sd-swim#readme",
  "dependencies": {
    "lodash": "^4.17.4",
    "minimist": "^1.2.0",
    "pino": "^4.5.3",
    "propagate": "^1.0.0",
    "protocol-buffers": "^3.2.1",
    "uuid": "^3.0.1"
  },
  "devDependencies": {
    "coveralls": "^2.13.1",
    "dependency-check": "^2.8.0",
    "eslint": "^3.19.0",
    "eslint-config-standard": "^10.2.1",
    "eslint-plugin-import": "^2.3.0",
    "eslint-plugin-node": "^5.0.0",
    "eslint-plugin-promise": "^3.5.0",
    "eslint-plugin-standard": "^3.0.1",
    "husky": "^0.13.4",
    "lab": "^13.1.0",
    "lab-espower-transformer": "^1.1.0",
    "nsp": "^2.6.3",
    "power-assert": "^1.4.3",
    "proxyquire": "^1.8.0"
  }
}


<|start_filename|>test/metadata.js<|end_filename|>
/* eslint no-console:0 */

const Lab = require('lab')
const lab = (exports.lab = Lab.script())
const { describe, it, beforeEach, afterEach } = lab
const SDSwim = require('../lib/sd-swim')
const pino = require('pino')
const assert = require('power-assert')
const { startNodes, stopNodes, delay } = require('./common')

describe('Metadata Protocol', () => {
  describe('given a started node with some data', () => {
    let target, node
    const nodeOpts = [{ port: 12340 }]
    const testEntries1 = [{ key: 'test', value: Buffer.from('test') }]
    const testEntries2 = [{ key: 'test2', value: Buffer.from('test2') }]
    const testEntries3 = [{ key: 'test3', value: Buffer.from('test3') }]

    beforeEach(() =>
      startNodes(nodeOpts).then(results => {
        ;[target] = results
      })
    )

    afterEach(() =>
      stopNodes([target, node]).then(() => {
        target.removeAllListeners('new-metadata') // To avoid listeners on not yet gc objects
        node.removeAllListeners('new-metadata')
      })
    )

    it('should start a node and receive the metadata, then then node adds his entries and check that is propagated', done => {
      const hosts = [{ host: '127.0.0.1', port: target.port }]

      // start a single node that join the target.
      node = new SDSwim({ logger: pino(), port: 12341, hosts })
      node.on('new-metadata', data => {
        assert.deepEqual(data[0].entries, testEntries1)
        assert.deepEqual(data[0].owner, {
          host: '127.0.0.1',
          port: nodeOpts[0].port
        })
        assert.deepEqual(data[0].version, 1)
        node.addAll(testEntries2)
      })

      target.on('new-metadata', data => {
        const expected = [
          {
            owner: { host: '127.0.0.1', port: 12341 },
            entries: [{ key: 'test2', value: Buffer.from('test2') }],
            version: 1
          },
          {
            owner: { host: '127.0.0.1', port: 12340 },
            version: 1,
            entries: [{ key: 'test', value: Buffer.from('test') }]
          }
        ]
        assert.deepEqual(target.data, expected)
        assert.deepEqual(data, expected)
        node.stop(() => {
          done()
        })
      })
      target.addAll(testEntries1)
      node.start()
    })

    it(
      'should start a node and receive the metadata, then new node add his and check that is propagated, ' +
        'then the original node will update his entries',
      done => {
        const hosts = [{ host: '127.0.0.1', port: target.port }]

        // start a single node that join the target.
        node = new SDSwim({ logger: pino(), port: 12341, hosts })

        node.on('new-metadata', data => {
          node.addAll(testEntries2)
        })

        target.on('new-metadata', data => {
          target.addAll(testEntries3) // add new node metadata
          node.on('new-metadata', data => {
            const expected = [
              {
                owner: { host: '127.0.0.1', port: 12340 },
                entries: [{ key: 'test3', value: Buffer.from('test3') }],
                version: 2
              },
              {
                owner: { host: '127.0.0.1', port: 12341 },
                version: 1,
                entries: [{ key: 'test2', value: Buffer.from('test2') }]
              }
            ]
            assert.deepEqual(data, expected)
            node.stop(() => {
              done()
            })
          })
        })
        target.addAll(testEntries1)
        node.start()
      }
    )

    it('should start a node and receive the metadata, then the node is stopped and metadata must be removed', done => {
      const hosts = [{ host: '127.0.0.1', port: target.port }]
      // start a single node that join the target.
      node = new SDSwim({ logger: pino(), port: 12341, hosts })
      node.on('new-metadata', data => {
        node.addAll(testEntries2)
      })

      let firstReceived = false
      target.on('new-metadata', data => {
        if (firstReceived) {
          return
        }
        firstReceived = true

        node.stop().then(delay(2000)).then(() => {
          const expected = [
            {
              owner: { host: '127.0.0.1', port: 12340 },
              entries: [{ key: 'test', value: Buffer.from('test') }],
              version: 1
            }
          ]
          assert.deepEqual(target.data, expected)
          done()
        })
      })
      target.addAll(testEntries1)
      node.start()
    })

    it('should start a node and receive the metadata, then new node add and remove new entry', done => {
      const hosts = [{ host: '127.0.0.1', port: target.port }]

      // start a single node that join the target.
      node = new SDSwim({ logger: pino(), port: 12341, hosts })

      node.on('up', data => {
        node.add('test2', Buffer.from('test2'))
      })

      let firstUpdate = true
      target.on('new-metadata', data => {
        if (firstUpdate) {
          const expected = [
            {
              owner: { host: '127.0.0.1', port: 12341 },
              entries: [
                {
                  key: 'test2',
                  value: Buffer.from('test2')
                }
              ],
              version: 1
            },
            {
              owner: { host: '127.0.0.1', port: 12340 },
              version: 0,
              entries: []
            }
          ]
          assert.deepEqual(data, expected)
          node.remove('test2')
          firstUpdate = false
        } else {
          const expected = [
            {
              owner: { host: '127.0.0.1', port: 12341 },
              entries: [],
              version: 2
            },
            {
              owner: { host: '127.0.0.1', port: 12340 },
              version: 0,
              entries: []
            }
          ]
          assert.deepEqual(data, expected)
          done()
        }
      })
      node.start()
    })
  })
})


<|start_filename|>test/net.js<|end_filename|>
/* eslint no-console:0 */

const assert = require('power-assert')
const Lab = require('lab')
const lab = exports.lab = Lab.script()
const pino = require('pino')
const proxyquire = require('proxyquire')
const {describe, it, before, beforeEach} = lab
const Messages = require('../lib/messages')
const Dissemination = require('../lib/dissemination')

describe('Network Communication', () => {
  let net
  let sentMessages = []
  const opts = {}
  const messages = new Messages(opts)

  before(done => {
    const mockedClient = {
      send: (message, offset, length, port, address, cb) => {
        sentMessages.push({message, offset, length, port, address})
        cb()
      },
      close: () => {}
    }
    const mockedDgram = {
      createSocket: () => (mockedClient)
    }
    const Net = proxyquire('../lib/net', {
      'dgram': mockedDgram
    })
    net = new Net({logger: pino()})
    net.dissemination = new Dissemination({})
    done()
  })

  beforeEach(done => {
    sentMessages = []
    done()
  })

  it('should send a bunch of join messages', done => {
    const host1 = {host: 'host1', port: 1234}
    const host2 = {host: 'host2', port: 5678}
    const hosts = [host1, host2]
    const joinMessages = messages.joinMessages(hosts)
    net._sendMessages(joinMessages)
    assert.equal(sentMessages.length, 2)
    const firstJoin = messages.decodeMessage(sentMessages[0].message)
    const secondJoin = messages.decodeMessage(sentMessages[1].message)

    assert.deepEqual(firstJoin.destination, host1)
    assert.equal(firstJoin.type, 0)

    assert.deepEqual(secondJoin.destination, host2)
    assert.equal(secondJoin.type, 0)

    done()
  })
})


<|start_filename|>lib/metadata.js<|end_filename|>
'use strict'

const { EventEmitter } = require('events')
const { find, remove, isBuffer, isString } = require('lodash')

// Metadata distribution subprotocol module.
//
// Each metadata entry must be in the form:
// {owner: {host, port}, entries: [{key, value}], version}
// The entries structure `[{key, value}]` is set by protobuf messages (this module doesen't care about the entry content)
//
// Logic:
// - When a node changes metadata, it sends that change to all nodes (not caring about acks), increasing the version
// - On `peerUp` the node sends his metadata to the new peer
// - On `peerDown` the failing node metadata are removed
// - Every `metadataDistributionTimeout`, a member is chosen and all the metadata are sent
// - If a `all-meta` message is received, all the metadata are returned to the sender
//
// - Events emitted:
// - `new-metadata`

class Metadata extends EventEmitter {
  constructor (opts) {
    super()
    this.opts = opts

    this.members = opts.members
    this.logger = opts.logger
    this.net = opts.net
    this.sdswim = opts.sdswim

    // Specific Metadata subprotocol options
    this.metadataDistributionTimeout = opts.metadataDistributionTimeout || 1000

    // Metadata
    this.all = [] // all metadata
    this.entries = [] // my metadata  [{key, value}]
    this.version = 0 // my metadata version

    // Events managed
    this.members.on('peerUp', node => this._notifyMyMetaToNode(node))
    this.members.on('peerDown', node => this._removeNodeMetadata(node))

    // Messages received
    this.net.on('meta', (sender, msg) => this._onMeta(sender, msg))
    this.net.on('all-meta', (sender, msg) => this._onAllMeta(sender, msg))

    // Notify my data to everyone, triggering only if I know myself
    this.notifyToAll = () => {
      if (this.sdswim.whoami().host) {
        return this._notifyMetaToAll([this.my])
      }
    }
  }

  start () {
    this.logger.debug('Starting Metadata Distrubtion sub-protocol')
    this.sendInterval = setInterval(
      this._send.bind(this),
      this.metadataDistributionTimeout
    )
  }

  stop () {
    clearInterval(this.sendInterval)
    this.all = []
    this.entries = []
    this.version = 0
  }

  /**
   * Get all the metadata, including mine.
   */
  get data () {
    return this.all.concat([this.my])
  }

  /**
   * Change my data.
   * // TODO: add validation? entries must be in the form [{key, value}]
   */
  set my (entries) {
    // Validation, throw an error if key/value is of the wrong type
    for (const { key, value } of entries) {
      if (!isString(key) || !isBuffer(value)) {
        throw new Error(
          `Wrong type for ${key}. Key must be a String, value must be a Buffer`
        )
      }
    }
    this.entries = entries
    this.version++
    this.notifyToAll()
  }

  /**
   * Add a single entry. If there's a key with the same value, it's replaced
   */
  add (key, value) {
    if (!isBuffer(value)) {
      throw new Error('value must be a Buffer')
    }
    remove(this.entries, el => el.key === key)
    this.entries.push({ key, value })
    this.version++
    this.notifyToAll()
  }

  /**
   * Remove a single entry
   */
  remove (key) {
    remove(this.entries, el => el.key === key)
    this.version++
    this.notifyToAll()
  }

  /**
   * Get my metadata.
   */
  get my () {
    return {
      owner: this.members.me(),
      version: this.version,
      entries: this.entries
    }
  }

  _notifyMetaToAll (metadata) {
    this.net.sendMeta(
      this.members.getOtherMembers().map(({ node }) => node),
      metadata
    )
  }

  _notifyMyMetaToNode (node) {
    this.net.sendMeta([node], [this.my])
  }

  _notifyAllMetaToNode (node) {
    this.net.sendMeta([node], this.data)
  }

  _removeNodeMetadata ({ host, port }) {
    this.all = this.all.filter(
      meta => !(meta.owner.host === host && meta.owner.port === port)
    )
    this.emit('new-metadata', this.data)
  }

  /**
   * RECEIVED META MESSAGE. Updates metadata if the version is >
   */
  _onMeta ({ sender: { host, port } }, msg) {
    this.logger.debug(
      `Received META message from ${host}:${port} of type: ${msg.type}`
    )
    let changes
    for (const updated of msg.metadata) {
      const meta = find(
        this.all,
        el => updated.owner.host === host && updated.owner.port === port
      )
      if (!meta) {
        this.all.push(updated)
        changes = true
      } else if (updated.version > meta.version) {
        meta.entries = updated.entries
        meta.version = updated.version
        changes = true
      }
    }
    if (changes) {
      this.emit('new-metadata', this.data)
    }
  }

  /**
   * RECEIVED ALL_META MESSAGE. Sends back all the metadata I know.
   */
  _onAllMeta ({ sender }, msg) {
    this.logger.debug(
      `Received META message from ${sender.host}:${sender.port} of type: ${msg.type}`
    )
    this._notifyAllMetaToNode(sender)
  }

  _send () {
    const member = this.members.getPingableMember()
    if (!member) {
      // no candidates
      return
    }
    this._notifyMyMetaToNode(member.node)
  }
}
module.exports = Metadata


<|start_filename|>lib/members.js<|end_filename|>
const { find, remove, shuffle } = require('lodash')
const { nodeStates: { ALIVE, FAULTY, SUSPECT } } = require('../lib/states')
const { EventEmitter } = require('events')

// Memebers list and operations on it
// Also, after an interval, move a SUSPECT node to FAULTY, remove the member and generate the updates
// Events:
// - updated-members: when the whole list is set after a join
// - peerUp
// - peerDown

class Members extends EventEmitter {
  constructor (opts) {
    super()
    this.opts = opts
    this.sdswim = opts.sdswim
    this.dissemination = opts.dissemination
    this.logger = this.opts.logger

    // Members list options
    this.suspectTimeout = opts.suspectTimeout || 1000

    this.members = [] // The actual member list. Contains all the current ALIVE and SUSPECT members
    this.last = 0 // Index of the "next" element returned

    this.getRandomMemberIndex = arr => Math.floor(Math.random() * arr.length)
  }

  get list () {
    return this.members
  }

  set list (list) {
    this.members = list
    this.emit('updated-members', this.members.map(({ node }) => node)) // only the nodes
  }

  /**
   * Add member to member list with a node and state
   * The node is updated if already in list or added if not.
   * Default: add himself as ALIVE, setBy as himself, incNumber = 0
   */
  addOrUpdateMember (
    node = this.sdswim.me,
    state = ALIVE,
    setBy = this.sdswim.me,
    incNumber = 0
  ) {
    const current = this.findMember(node)
    if (current) {
      current.setBy = setBy
      current.state = state
      current.incNumber = incNumber
    } else {
      // we add the element in a random position
      const index = this.getRandomMemberIndex(this.members)
      this.members.splice(index, 0, { node, setBy, state, incNumber })
      if (state === ALIVE) {
        // added ALIVE member, but emitting only if not myself
        if (this.isMe(node)) {
          return
        }
        this.emit('peerUp', node)
      }
    }
  }

  /**
   * Add member to member list when (same as addOrUpdateMember but propagate the updates).
   */
  addOrUpdateMemberWithPropagate (
    node = this.sdswim.me,
    state = ALIVE,
    setBy = this.sdswim.me,
    incNumber = 0
  ) {
    this.addOrUpdateMember.apply(this, arguments)
    this.dissemination.propagate({ node, setBy, state, incNumber })
  }

  /**
   * If SUSPECT updates to ALIVE. Do nothing otherwise.
   * Creates also the related updates
   */
  updateToAlive (member) {
    const current = this.findMember(member)
    if (!current) {
      return
    }
    const { node, state, incNumber } = current
    if (state === SUSPECT) {
      this.dissemination.addUpdate(node, this.sdswim.me, ALIVE, incNumber)
      this.logger.debug(node, 'updated to ALIVE FROM SUSPECT')
    }
  }

  /**
   * If ALIVE update to SUSPECT. Do nothing otherwise.
   * Creates also the related updates
   */
  updateToSuspect (member) {
    const current = this.findMember(member)
    if (!current) {
      return
    }
    const { node, state, incNumber } = current
    if (state === ALIVE) {
      current.state = SUSPECT

      this.dissemination.addUpdate(node, this.sdswim.me, SUSPECT, incNumber)

      // Suspect timeout: when triggered it check if the node is still SUSPECT.
      // If so, it's set to faulty.
      setTimeout(() => {
        const current = this.findMember(member)
        if (current && current.state === SUSPECT) {
          this.updateToFaulty(node, this.me(), incNumber)
        }
      }, this.suspectTimeout)

      this.logger.debug(node, 'updated to SUSPECT FROM ALIVE')
    }
  }

  /**
   * Update to FAULTY (remove the moemebr from the list). Do nothing otherwise.
   * Creates also the related updates
   */
  updateToFaulty (member, setBy, incNumber) {
    this.removeMember(member)
    this.dissemination.addUpdate(member, setBy, FAULTY, incNumber)
  }

  /**
   * Update to FAULTY (remove the member from the list). Do nothing otherwise.
   */
  removeMember (member) {
    const prevNum = this.members.length
    remove(
      this.members,
      el => el.node.host === member.host && el.node.port === member.port
    )
    if (prevNum !== this.members.length) {
      this.emit('peerDown', member) // emitted only if the member hasn't laready removed
    }
  }

  // Get next `n` members for the ping/ping-req. The members array is traversed
  // circularly  and if traversal is completed in the process, is then shuffled
  // `membersToSkip` are excluded
  getNextMembers (membersToSkip, n) {
    const ret = []
    if (n <= 0 || this.members.length === 0) {
      return ret
    }
    let curr = this.last
    let exit = false
    let shuffleWhenDone = false
    while (!exit) {
      const current = this.members[curr]
      if (current) {
        const skip = membersToSkip.some(
          ({ host, port }) =>
            current.node.host === host && current.node.port === port
        )
        if (!skip) {
          ret.push(current)
        }
        curr++
        if (curr === this.members.length) {
          curr = 0
          shuffleWhenDone = true
        }
        // We exit if n is reached or we are back to the element from which we started
        if (ret.length === n || curr === this.last) {
          exit = true
          this.last = curr
        }
      } else {
        this.last = 0
        exit = true
      }
    }
    // Shuffle if in the process above we reached the end of the array
    if (shuffleWhenDone) {
      this.members = shuffle(this.members)
      this.last = 0
    }
    return ret
  }

  /**
   * Get the member target for the PING.
   */
  getPingableMember () {
    const pingableMembers = this.getNextMembers([this.sdswim.me], 1)
    if (!pingableMembers.length) {
      return // no candidates
    }
    return pingableMembers[0]
  }

  /**
   * Get the `pingReqGroupSize` members target for the PING-REQ.
   */
  getPingReqGroupMembers (target, pingReqGroupSize) {
    return this.getNextMembers([this.sdswim.me, target], pingReqGroupSize)
  }

  /**
   * Get the member, returns `null` if not found
   */
  findMember ({ host, port }) {
    return find(
      this.members,
      el => el.node.host === host && el.node.port === port
    )
  }

  isMe ({ host, port }) {
    return host === this.sdswim.me.host && port === this.sdswim.me.port
  }

  me () {
    return { host: this.sdswim.me.host, port: this.sdswim.me.port }
  }

  size () {
    return this.members.length
  }

  /**
   * Get the member target for the PING.
   */
  getOtherMembers () {
    return this.members.filter(el => !this.isMe(el.node))
  }
}

module.exports = Members


<|start_filename|>test/common.js<|end_filename|>
const SDSwim = require('../lib/sd-swim')
const { deepEqual } = require('power-assert')
const { sortBy } = require('lodash')

const compareMember = member => `${member.node.host}:${member.node.port}`
const compareNode = node => `${node.host}:${node.port}`

exports.startNode = opts => {
  const target = new SDSwim(opts)
  return target.start().then(() => target)
}
exports.stopNode = node => (node ? node.stop() : null)
exports.startNodes = opts =>
  Promise.all(opts.map(opt => exports.startNode(opt)))
exports.stopNodes = nodes =>
  Promise.all(nodes.map(node => exports.stopNode(node)))
exports.delay = t => () => new Promise(resolve => setTimeout(resolve, t))
// exports.compareNodeLists = (expected, actual) => deepEqual(sortBy(expected, compareHost), sortBy(actual, compareHost))
exports.compareMemberLists = (expected, actual) =>
  deepEqual(sortBy(expected, [compareMember]), sortBy(actual, [compareMember]))
exports.compareNodesLists = (expected, actual) =>
  deepEqual(sortBy(expected, [compareNode]), sortBy(actual, [compareNode]))


<|start_filename|>test/members.js<|end_filename|>
const assert = require('power-assert')
const Lab = require('lab')
const {compareMemberLists} = require('./common')
const lab = exports.lab = Lab.script()
const {describe, it, beforeEach, afterEach} = lab
const SDSwim = require('../lib/sd-swim')
const {nodeStates: {ALIVE, SUSPECT, FAULTY}} = require('../lib/states')

describe('Members', () => {
  let node
  const nodePort = 12345

  beforeEach(done => {
    node = new SDSwim({port: nodePort})
    node.host = '1.1.1.0'
    node.start(done)
  })

  afterEach(done => {
    node.stop(done)
  })

  it('should update the member list correctly', done => {
    const members = node.members

    const host1 = {host: '1.1.1.1', port: 1111}
    const host2 = {host: '1.1.1.2', port: 1112}
    const host3 = {host: '1.1.1.3', port: 1112}
    const setBy = {host: node.host, port: node.port}

    const expected1 = {node: host1, state: ALIVE, setBy, incNumber: 0}
    const expected2 = {node: host2, state: ALIVE, setBy, incNumber: 0}
    const expected3 = {node: host1, state: FAULTY, setBy, incNumber: 0}
    const expected4 = {node: host1, state: SUSPECT, setBy: host3, incNumber: 0}

    assert.deepEqual([], members.list)
    members.addOrUpdateMember(host1, ALIVE)
    members.addOrUpdateMember(host2, ALIVE)
    compareMemberLists([expected1, expected2], members.list)

    members.addOrUpdateMember(host1, FAULTY)
    compareMemberLists([expected3, expected2], members.list)

    members.addOrUpdateMember(host1, SUSPECT, host3) // set by a different host
    compareMemberLists([expected4, expected2], members.list)

    done()
  })

  it('should get the correct ping member with two members', done => {
    const members = node.members
    const host1 = {host: '1.1.1.1', port: 1111}
    members.addOrUpdateMember(host1, ALIVE)
    const pingable = members.getPingableMember()
    assert.deepEqual(pingable.node, host1)
    done()
  })

  it('should get the correct ping member with tree members', done => {
    const members = node.members
    const host1 = {host: '1.1.1.1', port: 1111}
    const host2 = {host: '1.1.1.2', port: 1112}
    const me = members.me()
    members.addOrUpdateMember(host1, ALIVE)
    members.addOrUpdateMember(host2, ALIVE)
    const pingable = members.getPingableMember()
    assert.notDeepEqual(pingable.node, me)
    done()
  })

  it('should get the correct ping-req target whit zero group size', done => {
    const members = node.members
    const host1 = {host: '1.1.1.1', port: 1111}
    const host2 = {host: '1.1.1.2', port: 1112}
    const host3 = {host: '1.1.1.3', port: 1112}
    const host4 = {host: '1.1.1.4', port: 1112}
    members.addOrUpdateMember(host1, ALIVE)
    members.addOrUpdateMember(host2, ALIVE)
    members.addOrUpdateMember(host3, ALIVE)
    members.addOrUpdateMember(host4, ALIVE)
    members.addOrUpdateMember(host1, ALIVE)
    const pingables = members.getPingReqGroupMembers(host1, 0)
    assert.deepEqual(pingables, [])
    done()
  })

  it('should get the correct ping-req target whit no pingable candidates', done => {
    const members = node.members
    const host1 = {host: '1.1.1.1', port: 1111}
    members.addOrUpdateMember(host1, ALIVE)
    const pingables = members.getPingReqGroupMembers(host1, 10)
    assert.deepEqual(pingables, [])
    done()
  })

  it('should get the correct ping-req target with four members', done => {
    const members = node.members
    const host1 = {host: '1.1.1.1', port: 1111}
    const host2 = {host: '1.1.1.2', port: 1112}
    const host3 = {host: '1.1.1.3', port: 1112}
    const host4 = {host: '1.1.1.4', port: 1112}
    members.addOrUpdateMember(host1, ALIVE)
    members.addOrUpdateMember(host2, ALIVE)
    members.addOrUpdateMember(host3, ALIVE)
    members.addOrUpdateMember(host4, ALIVE)
    const pingables = members.getPingReqGroupMembers(host1, 3)
    assert(pingables.length, 3)
    done()
  })

  it('should get the correct ping-req target when reaching the end of members list', done => {
    const members = node.members
    const host1 = {host: '1.1.1.1', port: 1111}
    const host2 = {host: '1.1.1.2', port: 1112}
    const host3 = {host: '1.1.1.3', port: 1112}
    const host4 = {host: '1.1.1.4', port: 1112}
    const host5 = {host: '1.1.1.5', port: 1112}
    members.addOrUpdateMember(host1, ALIVE)
    members.addOrUpdateMember(host2, ALIVE)
    members.addOrUpdateMember(host3, ALIVE)
    members.addOrUpdateMember(host4, ALIVE)
    members.addOrUpdateMember(host5, ALIVE)
    const me = members.me()
    const pingables1 = members.getPingReqGroupMembers(host1, 3)
    assert(pingables1.length, 3)
    for (const node of pingables1) {
      if (node.host === host1.host || node.host === me.host) {
        assert.fail(`{node}shoudn't be returned`)
      }
    }
    const pingables2 = members.getPingReqGroupMembers(host1, 3)
    assert(pingables2.length, 3)
    for (const node of pingables2) {
      if (node.host === host1.host || node.host === me.host) {
        assert.fail(`{node}shoudn't be returned`)
      }
    }
    // check after the shuffle
    const pingables3 = members.getPingReqGroupMembers(host1, 3)
    assert(pingables3.length, 3)
    for (const node of pingables3) {
      if (node.host === host1.host || node.host === me.host) {
        assert.fail(`{node}shoudn't be returned`)
      }
    }
    done()
  })
})


<|start_filename|>test/join.js<|end_filename|>
/* eslint no-console:0 */

const assert = require('power-assert')
const Lab = require('lab')
const lab = exports.lab = Lab.script()

const {describe, it, beforeEach, afterEach} = lab
const SDSwim = require('../lib/sd-swim')
const {states: {JOINED}} = require('../lib/states')
const {compareNodesLists} = require('./common')

describe('Join', () => {
  let target
  const targetPort = 12345

  beforeEach(done => {
    target = new SDSwim({port: targetPort})
    target.start(done)
  })

  afterEach(done => {
    target.stop(done)
  })

  it('should join and emit the joined event', done => {
    const hosts = [{host: '127.0.0.1', port: target.port}]
    // start a single node that join the target.
    const sdswim = new SDSwim({port: 12340, hosts})
    sdswim.on('joined', () => {
      const myself = sdswim.whoami()
      assert.strictEqual(myself.state, JOINED)
      sdswim.stop(() => {
        done()
      })
    })
    sdswim.start()
  })

  it('should new node send a join message and get a correct member list', done => {
    const port = 12340
    const hosts = [{host: '127.0.0.1', port: targetPort}]
    const sdswim = new SDSwim({port, hosts})
    sdswim.start((err) => {
      assert.equal(err, null)
    })
    sdswim.on('updated-members', membersList => {
      const expectedList = [
        { host: '127.0.0.1', port: 12345 },
        { host: '127.0.0.1', port: 12340 }]
      compareNodesLists(membersList, expectedList)
      compareNodesLists(target.memberList, expectedList)

      assert.deepEqual(sdswim.host, '127.0.0.1')
      assert.deepEqual(sdswim.port, 12340)
      assert.deepEqual(target.host, '127.0.0.1')
      assert.deepEqual(target.port, 12345)

      sdswim.stop(done)
    })
  })

  it('should new node reach the join timeout when trying to join a non-existing node', done => {
    const port = 12340
    const hosts = [{host: '127.0.0.1', port: targetPort + 1}]
    const sdswim = new SDSwim({port, hosts, joinTimeout: 1000})
    sdswim.start((err) => {
      assert.equal(err, null)
    })
    sdswim.on('updated-members', () => {
      assert.fail('shoudn\'t receive an answer to join')
      sdswim.stop(done)
    })
    sdswim.on('error', err => {
      assert.ok(err.message.includes('Timeout triggered'))
      sdswim.stop(done)
    })
  })

  it.skip('TODO - MISSING TEST: join more than one host')
  it.skip('TODO - MISSING TEST: timout trigger when join successful')
})


<|start_filename|>index.js<|end_filename|>
const pino = require('pino')
const minimist = require('minimist')
const SDSwim = require('./lib/sd-swim')

/* eslint no-console:0 */

function start () {
  const logger = pino()
  const info = logger.info.bind(logger)
  const debug = logger.info.bind(logger)
  const error = logger.error.bind(logger)

  const argv = minimist(process.argv.slice(2), {
    integer: ['port', 'list'],
    boolean: ['verbose'],
    alias: {
      port: 'p',
      help: 'H',
      verbose: 'v',
      list: 'l',
      joinTimeout: 'j' // join timeout
    },
    default: {
      // We could also assume that the default is 0. If so, UDP try to bind to
      // a random port
      port: process.env.SWIM_PORT || 11000,
      verbose: false,
      list: 0
    }
  })

  const usage = `Usage: ${process.argv[1]} [-p PORT] [-jt 2000] [-l 1000] -v host1[:port1] host2[:port2]...`
  if (argv.help) {
    console.error(usage)
    process.exit(1)
  }

  const nodes = argv._ //  contains all the arguments that didn't have an option associated with them
  const hosts = nodes.map(el => {
    const [host, portStr] = el.split(':')
    const port = Number(portStr)
    if (!port || port < 0 || port > 65535) {
      console.error(`Port ${portStr} not correct, must be a port valid number`)
      console.error(usage)
      process.exit(1)
    }
    return {host, port}
  })

  if (argv.verbose) {
    logger.level = 'debug'
  }

  const opts = Object.assign({hosts, logger}, argv)
  opts.logLevel = 'debug'
  const sdswim = new SDSwim(opts)

  sdswim.on('join-sent', () => {
    info('join reques sent to nodes')
  })

  // nothing on generic error
  sdswim.on('up', port => {
    info({event: 'I am up', port})
  })

  sdswim.on('joined', () => {
    info({id: sdswim.whoami()}, 'Joined')
  })

  sdswim.on('updated-members', members => {
    debug({members}, 'Members List Update')
  })

  sdswim.on('error', err => {
    error(err)
    process.exit(1)
  })
  sdswim.start()

  if (argv.list) {
    setInterval(() => {
      console.log('*****************************************')
      console.log(sdswim.memberList)
    }, argv.list)
  }
}

if (require.main === module) {
  start()
}

module.exports = require('./lib/sd-swim')


<|start_filename|>test/dissemination.js<|end_filename|>
/* eslint no-console:0 */

const Lab = require('lab')
const lab = exports.lab = Lab.script()
const {omit} = require('lodash')
const assert = require('power-assert')
const SDSwim = require('../lib/sd-swim')
const {nodeStates: {ALIVE, SUSPECT, FAULTY}} = require('../lib/states')
const {describe, it, beforeEach, afterEach} = lab

describe('Dissemination', () => {
  let node
  beforeEach(done => {
    node = new SDSwim({port: 12345})
    node.start(done)
  })

  afterEach(done => {
    node.stop(done)
  })

  const host1 = {host: '127.0.0.1', port: 1234}
  const host2 = {host: '127.0.0.1', port: 5678}
  const host3 = {host: '127.0.0.1', port: 5679}
  const host4 = {host: '127.0.0.1', port: 5680}

  it('should create an alive update correctly', done => {
    const dissemination = node.opts.dissemination
    const expected = [{node: host1, setBy: host2, state: ALIVE, incNumber: 0, uuid: '1'}]
    dissemination.addUpdate(host1, host2, ALIVE, 0, '1')
    const updates = dissemination.getUpdates()
    assert.deepEqual(updates, expected)
    const temp = dissemination.getUpdates()
    assert.deepEqual([], temp) // no more updates
    done()
  })

  it('should create an alive update correctly passing the state', done => {
    const dissemination = node.opts.dissemination
    const expected = [{node: host1, setBy: host2, state: ALIVE, incNumber: 1, uuid: '1'}]
    dissemination.addUpdate(host1, host2, ALIVE, 1, '1')
    const updates = dissemination.getUpdates()
    assert.deepEqual(updates, expected)
    const temp = dissemination.getUpdates()
    assert.deepEqual(temp, []) // no more updates
    done()
  })

  it('should create a faulty update correctly', done => {
    const dissemination = node.opts.dissemination
    const expected = [{node: host1, setBy: host2, state: FAULTY, incNumber: 0, uuid: '1'}]
    dissemination.addUpdate(host1, host2, FAULTY, 0, '1')
    const updates = dissemination.getUpdates()
    assert.deepEqual(updates, expected)
    assert.deepEqual([], dissemination.getUpdates()) // no more updates
    done()
  })

  it('should create a suspect update correctly', done => {
    const dissemination = node.opts.dissemination
    const expected = [{node: host1, setBy: host2, state: SUSPECT, incNumber: 0, uuid: '1'}]
    dissemination.addUpdate(host1, host2, SUSPECT, 0, '1')
    const updates = dissemination.getUpdates()
    assert.deepEqual(updates, expected)
    assert.deepEqual(dissemination.getUpdates(), []) // no more updates
    done()
  })

  it('should create a list of updates <= updatesMaxSize, with no member list', done => {
    const dissemination = node.opts.dissemination
    const expected = [
      {node: host1, setBy: host2, state: SUSPECT, incNumber: 0, uuid: '1'},
      {node: host2, setBy: host2, state: FAULTY, incNumber: 0, uuid: '1'},
      {node: host3, setBy: host2, state: ALIVE, incNumber: 0, uuid: '1'},
      {node: host4, setBy: host2, state: ALIVE, incNumber: 0, uuid: '1'}
    ]
    dissemination.addUpdate(host1, host2, SUSPECT, 0, '1')
    dissemination.addUpdate(host2, host2, FAULTY, 0, '1')
    dissemination.addUpdate(host3, host2, ALIVE, 0, '1')
    dissemination.addUpdate(host4, host2, ALIVE, 0, '1')
    const updates = dissemination.getUpdates()
    assert.deepEqual(updates, expected)
    assert.deepEqual(dissemination.getUpdates(), []) // no more updates, member list is 0 so also the dissemination Limit
    done()
  })

  it('should create a list of updates <= updatesMaxSize, with member list', done => {
    const dissemination = node.opts.dissemination
    dissemination.opts.members.list = new Array(100) // a (empty) member list of 100 memebrs
    const expected = [
      {node: host1, setBy: host2, state: SUSPECT, incNumber: 0, uuid: '1'},
      {node: host2, setBy: host2, state: FAULTY, incNumber: 0, uuid: '1'},
      {node: host3, setBy: host2, state: ALIVE, incNumber: 0, uuid: '1'},
      {node: host4, setBy: host2, state: ALIVE, incNumber: 0, uuid: '1'}
    ]
    dissemination.addUpdate(host1, host2, SUSPECT, 0, '1')
    dissemination.addUpdate(host2, host2, FAULTY, 0, '1')
    dissemination.addUpdate(host3, host2, ALIVE, 0, '1')
    dissemination.addUpdate(host4, host2, ALIVE, 0, '1')
    const updates = dissemination.getUpdates()
    assert.deepEqual(updates, expected)
    assert.deepEqual(dissemination.getUpdates(), expected) // expecting the same
    done()
  })

  it('should create a list of updates > updatesMaxSize and return it correctly', done => {
    const dissemination = node.opts.dissemination
    dissemination.updatesMaxSize = 3
    const expected = [
      {node: host1, setBy: host2, state: SUSPECT, incNumber: 0, uuid: '1'},
      {node: host2, setBy: host2, state: FAULTY, incNumber: 0, uuid: '1'},
      {node: host3, setBy: host2, state: ALIVE, incNumber: 0, uuid: '1'}
    ]
    dissemination.addUpdate(host1, host2, SUSPECT, 0, '1')
    dissemination.addUpdate(host2, host2, FAULTY, 0, '1')
    dissemination.addUpdate(host3, host2, ALIVE, 0, '1')
    dissemination.addUpdate(host4, host2, ALIVE, 0, '1')
    const updates = dissemination.getUpdates()
    assert.deepEqual(updates, expected)
    done()
  })

  describe('given an ALIVE update', () => {
    it('should process an ALIVE update correctly with unknown member', done => {
      const dissemination = node.opts.dissemination
      const updateToAlive = {
        node: host1,
        state: ALIVE,
        setBy: host2,
        incNumber: 0,
        uuid: '1'
      }
      dissemination.processUpdates([updateToAlive])
      assert.deepEqual(node.memberList.map(el => omit(el, 'uuid')), [host1])
      assert.deepEqual(dissemination.getUpdates(), [updateToAlive])
      done()
    })

    it('should process an ALIVE update correctly with member with incNumber <= than the update', done => {
      const dissemination = node.opts.dissemination
      const updateToAlive = {node: host1, state: ALIVE, setBy: host2, incNumber: 0, uuid: '1'}
      dissemination.processUpdates([updateToAlive])
      assert.deepEqual(dissemination.getUpdates(), [updateToAlive])
      const newUpdateToAlive = {node: host1, state: ALIVE, setBy: host3, incNumber: 1, uuid: '1'}
      dissemination.processUpdates([newUpdateToAlive])
      assert.deepEqual(node.memberList.map(el => omit(el, 'uuid')), [host1])
      assert.deepEqual(dissemination.getUpdates(), [newUpdateToAlive, updateToAlive])
      done()
    })

    it('should process an ALIVE update correctly with member with incNumber > than the update', done => {
      const dissemination = node.opts.dissemination
      const updateToAlive = {node: host1, state: ALIVE, setBy: host2, incNumber: 1, uuid: '1'}
      dissemination.processUpdates([updateToAlive])
      const newUpdateToAlive = {node: host1, state: ALIVE, setBy: host3, incNumber: 0, uuid: '1'}
      dissemination.processUpdates([newUpdateToAlive])
      assert.deepEqual(node.memberList.map(el => omit(el, 'uuid')), [host1])
      assert.deepEqual(dissemination.getUpdates(), [updateToAlive])
      done()
    })

    it('should process an ALIVE update correctly with a SUSPECT member with incNumber <= than the update', done => {
      const dissemination = node.opts.dissemination
      const updateToSuspect = {node: host1, state: SUSPECT, setBy: host2, incNumber: 0, uuid: '1'}
      dissemination.processUpdates([updateToSuspect])
      const newUpdateToAlive = {node: host1, state: ALIVE, setBy: host3, incNumber: 1, uuid: '1'}
      dissemination.processUpdates([newUpdateToAlive])
      assert.deepEqual(node.memberList.map(el => omit(el, 'uuid')), [host1])
      done()
    })
  })

  describe('given an SUSPECT update', () => {
    it('should process an SUSPECT update correctly when target is ME', done => {
      // a new ALIVE message must be produces with incNumber incremented
      const dissemination = node.opts.dissemination
      node.host = host1.host
      node.port = host1.port
      const updateToSuspect = {
        node: host1,
        state: SUSPECT,
        setBy: host2,
        incNumber: 0,
        uuid: '1'
      }
      dissemination.processUpdates([updateToSuspect])
      assert.deepEqual(node.memberList.map(el => omit(el, 'uuid')), [host1])

      const expectedMember = {
        node: host1,
        state: ALIVE,
        setBy: host1,
        incNumber: 1
      }
      assert.deepEqual(dissemination.getUpdates().map(el => omit(el, 'uuid')), [expectedMember])
      done()
    })

    it('should process an SUSPECT update correctly with unknown member', done => {
      const dissemination = node.opts.dissemination
      const updateToSuspect = {
        node: host1,
        state: SUSPECT,
        setBy: host2,
        incNumber: 0,
        uuid: '1'
      }
      dissemination.processUpdates([updateToSuspect])
      assert.deepEqual(node.memberList.map(el => omit(el, 'uuid')), [host1])
      assert.deepEqual(dissemination.getUpdates(), [updateToSuspect])
      done()
    })

    it('should process an SUSPECT update correctly with a ALIVE member with incNumber <= than the update', done => {
      const dissemination = node.opts.dissemination
      const updateToAlive = {node: host1, state: ALIVE, setBy: host2, incNumber: 0, uuid: '1'}
      dissemination.processUpdates([updateToAlive])
      assert.deepEqual(dissemination.getUpdates(), [updateToAlive])

      const newUpdateToSuspect = {node: host1, state: SUSPECT, setBy: host3, incNumber: 1, uuid: '1'}
      dissemination.processUpdates([newUpdateToSuspect])
      assert.deepEqual(node.memberList.map(el => omit(el, 'uuid')), [host1])
      assert.deepEqual(dissemination.getUpdates(), [newUpdateToSuspect, updateToAlive])
      done()
    })

    it('should ignore a SUSPECT update when ALIVE member with incNumber > than the update', done => {
      const dissemination = node.opts.dissemination
      const updateToAlive = {node: host1, state: ALIVE, setBy: host2, incNumber: 1, uuid: '1'}
      dissemination.processUpdates([updateToAlive])
      assert.deepEqual(dissemination.getUpdates(), [updateToAlive])

      const newUpdateToSuspect = {node: host1, state: SUSPECT, setBy: host3, incNumber: 0, uuid: '1'}
      dissemination.processUpdates([newUpdateToSuspect])
      assert.deepEqual(node.memberList.map(el => omit(el, 'uuid')), [host1])
      assert.deepEqual(dissemination.getUpdates(), [updateToAlive])
      done()
    })

    it('should process a SUSPECT update correctly with a SUSPECT member with incNumber <= than the update', done => {
      const dissemination = node.opts.dissemination
      const updateToSuspect = {node: host1, state: SUSPECT, setBy: host2, incNumber: 0, uuid: '1'}
      dissemination.processUpdates([updateToSuspect])
      assert.deepEqual(dissemination.getUpdates(), [updateToSuspect])

      const newUpdateToSuspect = {node: host1, state: SUSPECT, setBy: host3, incNumber: 1, uuid: '1'}
      dissemination.processUpdates([newUpdateToSuspect])
      assert.deepEqual(node.memberList.map(el => omit(el, 'uuid')), [host1])
      assert.deepEqual(dissemination.getUpdates(), [newUpdateToSuspect, updateToSuspect])
      done()
    })

    it('should ignore a SUSPECT update correctly with a SUSPECT member with incNumber > than the update', done => {
      const dissemination = node.opts.dissemination
      const updateToSuspect = {node: host1, state: SUSPECT, setBy: host2, incNumber: 1, uuid: '1'}
      dissemination.processUpdates([updateToSuspect])
      assert.deepEqual(dissemination.getUpdates(), [updateToSuspect])

      const newUpdateToSuspect = {node: host1, state: SUSPECT, setBy: host3, incNumber: 0, uuid: '1'}
      dissemination.processUpdates([newUpdateToSuspect])
      assert.deepEqual(node.memberList.map(el => omit(el, 'uuid')), [host1])
      assert.deepEqual(dissemination.getUpdates(), [updateToSuspect])
      done()
    })
  })

  describe('given an FAULTY update', () => {
    it('should ignore a FAULTY update if member is not present', done => {
      const dissemination = node.opts.dissemination
      const updateToFaulty = {
        node: host1,
        state: FAULTY,
        setBy: host2,
        incNumber: 0,
        uuid: '1'
      }
      dissemination.processUpdates([updateToFaulty])
      assert.deepEqual(node.memberList.map(el => omit(el, 'uuid')), [])
      done()
    })

    it('should create an ALIVE message if target is ME', done => {
      // a new ALIVE message must be produces with incNumber incremented
      const dissemination = node.opts.dissemination
      node.host = host1.host
      node.port = host1.port
      const updateToFaulty = {
        node: host1,
        state: FAULTY,
        setBy: host2,
        incNumber: 0,
        uuid: '1'
      }
      dissemination.processUpdates([updateToFaulty])
      assert.deepEqual(node.memberList.map(el => omit(el, 'uuid')), [host1])

      const expectedMember = {
        node: host1,
        state: ALIVE,
        setBy: host1,
        incNumber: 1
      }
      assert.deepEqual(dissemination.getUpdates().map(el => omit(el, 'uuid')), [expectedMember])
      done()
    })

    it('should create a FAULTY update correctly (remove the member and propagates the update) when member with incNumber < than the update', done => {
      const dissemination = node.opts.dissemination
      const updateToAlive = {node: host1, state: ALIVE, setBy: host2, incNumber: 0, uuid: '1'}
      dissemination.processUpdates([updateToAlive])
      assert.deepEqual(dissemination.getUpdates(), [updateToAlive])
      const newUpdateToFaulty = {node: host1, state: FAULTY, setBy: host3, incNumber: 1, uuid: '1'}
      dissemination.processUpdates([newUpdateToFaulty])
      assert.deepEqual(node.memberList.map(el => omit(el, 'uuid')), [])
      assert.deepEqual(dissemination.getUpdates(), [newUpdateToFaulty, updateToAlive])
      done()
    })
  })
})


<|start_filename|>lib/states.js<|end_filename|>
// Constants for my states and node states

exports.states = {
  STARTED: 'STARTED',
  JOINED: 'JOINED',
  STOPPED: 'STOPPED'
}

exports.nodeStates = {
  ALIVE: 0,
  SUSPECT: 1,
  FAULTY: 2
}


<|start_filename|>lib/join.js<|end_filename|>
'use strict'

// Join Protocol Implementation
// Events explicitely emitted:
// - joined: when join protocols succeded, no args
// - join-timeout: when join timeout occurs

const {EventEmitter} = require('events')
const {states: {JOINED}} = require('./states')

class Join extends EventEmitter {
  constructor (opts) {
    super()
    this.opts = opts
    this.sdswim = opts.sdswim

    // Needed objects
    this.net = this.opts.net
    this.members = this.opts.members
    this.logger = this.opts.logger

    // Specific JOIN Protocol options
    this.initialHosts = this.opts.initialHosts
    this.joinTimeout = this.opts.joinTimeout

    // NET messages managed by this module
    this.net.on('join', (sender, msg) => this._joinReceived(sender, msg))
    this.net.on('join-ack', (sender, msg) => this._joinAckReceived(sender, msg))
  }

  start () {
    this._join() // Starts the protocol
  }

  stop () {
    if (this.currentTimer) {
      clearTimeout(this.currentTimer)
    }
  }

  /**
   * SEND JOIN MESSAGE(S) TO EACH `initialHosts`
   */
  _join () {
    const self = this
    if (this.initialHosts && this.initialHosts.length > 0) {
      this.net.sendJoin(this.initialHosts)
      this.currentTimer = setTimeout(function checkJoinTimeout () {
        if (self.sdswim.state !== JOINED) {
          self.emit('join-timeout', new Error(`Timeout triggered in joining with ${JSON.stringify(self.initialHosts)}`))
        }
      }, this.joinTimeout)
    }
  }

  /**
   * RECEIVED JOIN MESSAGE.
   */
  _joinReceived ({sender: {host, port}}, msg) {
    this.logger.debug(`Received JOIN message from ${host}:${port} of type: ${msg.type}`)
    // if I don't know my host yet, saves it and add myself to the current member list
    if (!this.sdswim.host) {
      this.sdswim.host = msg.destination.host
      this.members.addOrUpdateMemberWithPropagate()
    }
    const destination = {host, port} // The new node
    this.members.addOrUpdateMemberWithPropagate(destination)
    this.net.sendJoinAckMessage(destination, this.members.list)
  }

  /**
   * RECEIVE JOIN_ACK MESSAGE. In general we will receive more tha one message. Only one is
   * enough to change the state to JOINED, but we process the answer updating the members list
   * list for each message.
   */
  _joinAckReceived ({sender: {host, port}}, msg) {
    this.logger.debug(`Received JOIN_ACK message from ${host}:${port} of type: ${msg.type}`)
    // We process only first join ack, the others are silently ignored
    if (!this.sdswim.host) {   // First join ack received
      this.sdswim.host = msg.destination.host
      this.members.list = msg.members // The full list is replaced by the new one.
      this.emit('joined')
    }
  }
}
module.exports = Join
