CloudWeb_YZ/public/mqtt/test/abstract_client.js

'use strict'

/**
 * Testing dependencies
 */
var should = require('should')
var sinon = require('sinon')
var mqtt = require('../')
var xtend = require('xtend')
var Server = require('./server')
var Store = require('./../lib/store')
var port = 9876

module.exports = function (server, config) {
  function connect (opts) {
    opts = xtend(config, opts)
    return mqtt.connect(opts)
  }

  describe('closing', function () {
    it('should emit close if stream closes', function (done) {
      var client = connect()

      client.once('connect', function () {
        client.stream.end()
      })
      client.once('close', function () {
        client.end()
        done()
      })
    })

    it('should mark the client as disconnected', function (done) {
      var client = connect()

      client.once('close', function () {
        client.end()
        if (!client.connected) {
          done()
        } else {
          done(new Error('Not marked as disconnected'))
        }
      })
      client.once('connect', function () {
        client.stream.end()
      })
    })

    it('should stop ping timer if stream closes', function (done) {
      var client = connect()

      client.once('close', function () {
        should.not.exist(client.pingTimer)
        client.end()
        done()
      })

      client.once('connect', function () {
        should.exist(client.pingTimer)
        client.stream.end()
      })
    })

    it('should emit close after end called', function (done) {
      var client = connect()

      client.once('close', function () {
        done()
      })

      client.once('connect', function () {
        client.end()
      })
    })

    it('should emit end after end called and client must be disconnected', function (done) {
      var client = connect()

      client.once('end', function () {
        if (client.disconnected) {
          return done()
        }
        done(new Error('client must be disconnected'))
      })

      client.once('connect', function () {
        client.end()
      })
    })

    it('should pass store close error to end callback but not to end listeners', function (done) {
      var store = new Store()
      var client = connect({outgoingStore: store})

      store.close = function (cb) {
        cb(new Error('test'))
      }
      client.once('end', function () {
        if (arguments.length === 0) {
          return done()
        }
        throw new Error('no argument shoould be passed to event')
      })

      client.once('connect', function () {
        client.end(function (test) {
          if (test && test.message === 'test') {
            return
          }
          throw new Error('bad argument passed to callback')
        })
      })
    })

    it('should return `this` if end called twice', function (done) {
      var client = connect()

      client.once('connect', function () {
        client.end()
        var value = client.end()
        if (value === client) {
          done()
        } else {
          done(new Error('Not returning client.'))
        }
      })
    })

    it('should emit end only on first client end', function (done) {
      var client = connect()

      client.once('end', function () {
        var timeout = setTimeout(done.bind(null), 200)
        client.once('end', function () {
          clearTimeout(timeout)
          done(new Error('end was emitted twice'))
        })
        client.end()
      })

      client.once('connect', client.end.bind(client))
    })

    it('should stop ping timer after end called', function (done) {
      var client = connect()

      client.once('connect', function () {
        should.exist(client.pingTimer)
        client.end()
        should.not.exist(client.pingTimer)
        done()
      })
    })

    it('should be able to end even on a failed connection', function (done) {
      var client = connect({host: 'this_hostname_should_not_exist'})

      var timeout = setTimeout(function () {
        done(new Error('Failed to end a disconnected client'))
      }, 500)

      setTimeout(function () {
        client.end(function () {
          clearTimeout(timeout)
          done()
        })
      }, 200)
    })

    it('should emit end even on a failed connection', function (done) {
      var client = connect({host: 'this_hostname_should_not_exist'})

      var timeout = setTimeout(function () {
        done(new Error('Disconnected client has failed to emit end'))
      }, 500)

      client.once('end', function () {
        clearTimeout(timeout)
        done()
      })

      setTimeout(client.end.bind(client), 200)
    })

    it('should emit end only once for a reconnecting client', function (done) {
      var client = connect({host: 'this_hostname_should_not_exist', connectTimeout: 10, reconnectPeriod: 10})

      client.once('end', function () {
        var timeout = setTimeout(done.bind(null))
        client.once('end', function () {
          clearTimeout(timeout)
          done(new Error('end emitted twice'))
        })
      })

      setTimeout(client.end.bind(client), 300)
    })
  })

  describe('connecting', function () {
    it('should connect to the broker', function (done) {
      var client = connect()
      client.on('error', done)

      server.once('client', function () {
        client.end()
        done()
      })
    })

    it('should send a default client id', function (done) {
      var client = connect()
      client.on('error', done)

      server.once('client', function (serverClient) {
        serverClient.once('connect', function (packet) {
          packet.clientId.should.match(/mqttjs.*/)
          serverClient.disconnect()
          done()
        })
      })
    })

    it('should send be clean by default', function (done) {
      var client = connect()
      client.on('error', done)

      server.once('client', function (serverClient) {
        serverClient.once('connect', function (packet) {
          packet.clean.should.be.true()
          serverClient.disconnect()
          done()
        })
      })
    })

    it('should connect with the given client id', function (done) {
      var client = connect({clientId: 'testclient'})
      client.on('error', function (err) {
        throw err
      })

      server.once('client', function (serverClient) {
        serverClient.once('connect', function (packet) {
          packet.clientId.should.match(/testclient/)
          serverClient.disconnect()
          done()
        })
      })
    })

    it('should connect with the client id and unclean state', function (done) {
      var client = connect({clientId: 'testclient', clean: false})
      client.on('error', function (err) {
        throw err
      })

      server.once('client', function (serverClient) {
        serverClient.once('connect', function (packet) {
          packet.clientId.should.match(/testclient/)
          packet.clean.should.be.false()
          serverClient.disconnect()
          done()
        })
      })
    })

    it('should require a clientId with clean=false', function (done) {
      try {
        var client = connect({ clean: false })
        client.on('error', function (err) {
          done(err)
          // done(new Error('should have thrown'));
        })
      } catch (err) {
        done()
      }
    })

    it('should default to localhost', function (done) {
      var client = connect({clientId: 'testclient'})
      client.on('error', function (err) {
        throw err
      })

      server.once('client', function (serverClient) {
        serverClient.once('connect', function (packet) {
          packet.clientId.should.match(/testclient/)
          serverClient.disconnect()
          done()
        })
      })
    })

    it('should emit connect', function (done) {
      var client = connect()
      client.once('connect', function () {
        client.end()
        done()
      })
      client.once('error', done)
    })

    it('should provide connack packet with connect event', function (done) {
      server.once('client', function (serverClient) {
        serverClient.connack({returnCode: 0, sessionPresent: true})

        server.once('client', function (serverClient) {
          serverClient.connack({returnCode: 0, sessionPresent: false})
        })
      })

      var client = connect()
      client.once('connect', function (packet) {
        should(packet.sessionPresent).be.equal(true)
        client.once('connect', function (packet) {
          should(packet.sessionPresent).be.equal(false)
          client.end()
          done()
        })
      })
    })

    it('should mark the client as connected', function (done) {
      var client = connect()
      client.once('connect', function () {
        client.end()
        if (client.connected) {
          done()
        } else {
          done(new Error('Not marked as connected'))
        }
      })
    })

    it('should emit error', function (done) {
      var client = connect({clientId: 'invalid'})
      client.once('connect', function () {
        done(new Error('Should not emit connect'))
      })
      client.once('error', function (error) {
        should(error.code).be.equal(2) // code for clientID identifer rejected
        client.end()
        done()
      })
    })

    it('should have different client ids', function (done) {
      var client1 = connect()
      var client2 = connect()

      client1.options.clientId.should.not.equal(client2.options.clientId)
      client1.end(true)
      client2.end(true)
      setImmediate(done)
    })
  })

  describe('handling offline states', function () {
    it('should emit offline events once when the client transitions from connected states to disconnected ones', function (done) {
      var client = connect({reconnectPeriod: 20})

      client.on('connect', function () {
        this.stream.end()
      })

      client.on('offline', function () {
        client.end(true, done)
      })
    })

    it('should emit offline events once when the client (at first) can NOT connect to servers', function (done) {
      // fake a port
      var client = connect({ reconnectPeriod: 20, port: 4557 })

      client.on('offline', function () {
        client.end(true, done)
      })
    })
  })

  describe('topic validations when subscribing', function () {
    it('should be ok for well-formated topics', function (done) {
      var client = connect()
      client.subscribe(
        [
          '+', '+/event', 'event/+', '#', 'event/#', 'system/event/+',
          'system/+/event', 'system/registry/event/#', 'system/+/event/#',
          'system/registry/event/new_device', 'system/+/+/new_device'
        ],
        function (err) {
          client.end(function () {
            if (err) {
              return done(new Error(err))
            }
            done()
          })
        }
      )
    })

    it('should return an error (via callbacks) for topic #/event', function (done) {
      var client = connect()
      client.subscribe(['#/event', 'event#', 'event+'], function (err) {
        client.end(false, function () {
          if (err) {
            return done()
          }
          done(new Error('Validations do NOT work'))
        })
      })
    })

    it('should return an empty array for duplicate subs', function (done) {
      var client = connect()
      client.subscribe('event', function (err, granted1) {
        if (err) {
          return done(err)
        }
        client.subscribe('event', function (err, granted2) {
          if (err) {
            return done(err)
          }
          granted2.should.Array()
          granted2.should.be.empty()
          done()
        })
      })
    })

    it('should return an error (via callbacks) for topic #/event', function (done) {
      var client = connect()
      client.subscribe('#/event', function (err) {
        client.end(function () {
          if (err) {
            return done()
          }
          done(new Error('Validations do NOT work'))
        })
      })
    })

    it('should return an error (via callbacks) for topic event#', function (done) {
      var client = connect()
      client.subscribe('event#', function (err) {
        client.end(function () {
          if (err) {
            return done()
          }
          done(new Error('Validations do NOT work'))
        })
      })
    })

    it('should return an error (via callbacks) for topic system/#/event', function (done) {
      var client = connect()
      client.subscribe('system/#/event', function (err) {
        client.end(function () {
          if (err) {
            return done()
          }
          done(new Error('Validations do NOT work'))
        })
      })
    })

    it('should return an error (via callbacks) for empty topic list', function (done) {
      var client = connect()
      client.subscribe([], function (err) {
        client.end()
        if (err) {
          return done()
        }
        done(new Error('Validations do NOT work'))
      })
    })

    it('should return an error (via callbacks) for topic system/+/#/event', function (done) {
      var client = connect()
      client.subscribe('system/+/#/event', function (err) {
        client.end(true, function () {
          if (err) {
            return done()
          }
          done(new Error('Validations do NOT work'))
        })
      })
    })
  })

  describe('offline messages', function () {
    it('should queue message until connected', function (done) {
      var client = connect()

      client.publish('test', 'test')
      client.subscribe('test')
      client.unsubscribe('test')
      client.queue.length.should.equal(3)

      client.once('connect', function () {
        client.queue.length.should.equal(0)
        setTimeout(function () {
          client.end(true, done)
        }, 10)
      })
    })

    it('should not queue qos 0 messages if queueQoSZero is false', function (done) {
      var client = connect({queueQoSZero: false})

      client.publish('test', 'test', {qos: 0})
      client.queue.length.should.equal(0)
      client.on('connect', function () {
        setTimeout(function () {
          client.end(true, done)
        }, 10)
      })
    })

    it('should queue qos != 0 messages', function (done) {
      var client = connect({queueQoSZero: false})

      client.publish('test', 'test', {qos: 1})
      client.publish('test', 'test', {qos: 2})
      client.subscribe('test')
      client.unsubscribe('test')
      client.queue.length.should.equal(2)
      client.on('connect', function () {
        setTimeout(function () {
          client.end(true, done)
        }, 10)
      })
    })

    it('should call cb if an outgoing QoS 0 message is not sent', function (done) {
      var client = connect({queueQoSZero: false})
      var called = false

      client.publish('test', 'test', {qos: 0}, function () {
        called = true
      })

      client.on('connect', function () {
        called.should.equal(true)
        setTimeout(function () {
          client.end(true, done)
        }, 10)
      })
    })

    it('should delay ending up until all inflight messages are delivered', function (done) {
      var client = connect()
      var subscribeCalled = false

      client.on('connect', function () {
        client.subscribe('test', function () {
          subscribeCalled = true
        })
        client.publish('test', 'test', function () {
          client.end(false, function () {
            subscribeCalled.should.be.equal(true)
            done()
          })
        })
      })
    })

    it('wait QoS 1 publish messages', function (done) {
      var client = connect()
      var messageReceived = false

      client.on('connect', function () {
        client.subscribe('test')
        client.publish('test', 'test', { qos: 1 }, function () {
          client.end(false, function () {
            messageReceived.should.equal(true)
            done()
          })
        })
        client.on('message', function () {
          messageReceived = true
        })
      })

      server.once('client', function (serverClient) {
        serverClient.on('subscribe', function () {
          serverClient.on('publish', function (packet) {
            serverClient.publish(packet)
          })
        })
      })
    })

    it('does not wait acks when force-closing', function (done) {
      // non-running broker
      var client = connect('mqtt://localhost:8993')
      client.publish('test', 'test', { qos: 1 })
      client.end(true, done)
    })

    it('should call cb if store.put fails', function (done) {
      const store = new Store()
      store.put = function (packet, cb) {
        process.nextTick(cb, new Error('oops there is an error'))
      }
      var client = connect({ incomingStore: store, outgoingStore: store })
      client.publish('test', 'test', { qos: 2 }, function (err) {
        if (err) {
          client.end(true, done)
        }
      })
    })
  })

  describe('publishing', function () {
    it('should publish a message (offline)', function (done) {
      var client = connect()
      var payload = 'test'
      var topic = 'test'

      client.publish(topic, payload)

      server.on('client', onClient)

      function onClient (serverClient) {
        serverClient.once('connect', function () {
          server.removeListener('client', onClient)
        })

        serverClient.once('publish', function (packet) {
          packet.topic.should.equal(topic)
          packet.payload.toString().should.equal(payload)
          packet.qos.should.equal(0)
          packet.retain.should.equal(false)
          client.end(true, done)
        })
      }
    })

    it('should publish a message (online)', function (done) {
      var client = connect()
      var payload = 'test'
      var topic = 'test'

      client.on('connect', function () {
        client.publish(topic, payload)
      })

      server.once('client', function (serverClient) {
        serverClient.once('publish', function (packet) {
          packet.topic.should.equal(topic)
          packet.payload.toString().should.equal(payload)
          packet.qos.should.equal(0)
          packet.retain.should.equal(false)
          client.end()
          done()
        })
      })
    })

    it('should publish a message (retain, offline)', function (done) {
      var client = connect({ queueQoSZero: true })
      var payload = 'test'
      var topic = 'test'
      var called = false

      client.publish(topic, payload, { retain: true }, function () {
        called = true
      })

      server.once('client', function (serverClient) {
        serverClient.once('publish', function (packet) {
          packet.topic.should.equal(topic)
          packet.payload.toString().should.equal(payload)
          packet.qos.should.equal(0)
          packet.retain.should.equal(true)
          called.should.equal(true)
          client.end()
          done()
        })
      })
    })

    it('should emit a packetsend event', function (done) {
      var client = connect()
      var payload = 'test_payload'
      var testTopic = 'testTopic'

      client.on('packetsend', function (packet) {
        if (packet.cmd === 'publish') {
          packet.qos.should.equal(0)
          packet.topic.should.equal(testTopic)
          packet.payload.should.equal(payload)
          packet.retain.should.equal(false)
          client.end()
          done()
        }
      })

      client.publish(testTopic, payload)
    })

    it('should accept options', function (done) {
      var client = connect()
      var payload = 'test'
      var topic = 'test'
      var opts = {
        retain: true,
        qos: 1
      }

      client.once('connect', function () {
        client.publish(topic, payload, opts)
      })

      server.once('client', function (serverClient) {
        serverClient.once('publish', function (packet) {
          packet.topic.should.equal(topic)
          packet.payload.toString().should.equal(payload)
          packet.qos.should.equal(opts.qos, 'incorrect qos')
          packet.retain.should.equal(opts.retain, 'incorrect ret')
          packet.dup.should.equal(false, 'incorrect dup')
          client.end()
          done()
        })
      })
    })

    it('should publish with the default options for an empty parameter', function (done) {
      var client = connect()
      var payload = 'test'
      var topic = 'test'
      var defaultOpts = {qos: 0, retain: false, dup: false}

      client.once('connect', function () {
        client.publish(topic, payload, {})
      })

      server.once('client', function (serverClient) {
        serverClient.once('publish', function (packet) {
          packet.topic.should.equal(topic)
          packet.payload.toString().should.equal(payload)
          packet.qos.should.equal(defaultOpts.qos, 'incorrect qos')
          packet.retain.should.equal(defaultOpts.retain, 'incorrect ret')
          packet.dup.should.equal(defaultOpts.dup, 'incorrect dup')
          client.end()
          done()
        })
      })
    })

    it('should mark a message as  duplicate when "dup" option is set', function (done) {
      var client = connect()
      var payload = 'duplicated-test'
      var topic = 'test'
      var opts = {
        retain: true,
        qos: 1,
        dup: true
      }

      client.once('connect', function () {
        client.publish(topic, payload, opts)
      })

      server.once('client', function (serverClient) {
        serverClient.once('publish', function (packet) {
          packet.topic.should.equal(topic)
          packet.payload.toString().should.equal(payload)
          packet.dup.should.equal(opts.dup, 'incorrect dup')
          client.end()
          done()
        })
      })
    })

    it('should fire a callback (qos 0)', function (done) {
      var client = connect()

      client.once('connect', function () {
        client.publish('a', 'b', function () {
          client.end()
          done()
        })
      })
    })

    it('should fire a callback (qos 1)', function (done) {
      var client = connect()
      var opts = { qos: 1 }

      client.once('connect', function () {
        client.publish('a', 'b', opts, function () {
          client.end()
          done()
        })
      })
    })

    it('should fire a callback (qos 2)', function (done) {
      var client = connect()
      var opts = { qos: 2 }

      client.once('connect', function () {
        client.publish('a', 'b', opts, function () {
          client.end()
          done()
        })
      })
    })

    it('should support UTF-8 characters in topic', function (done) {
      var client = connect()

      client.once('connect', function () {
        client.publish('中国', 'hello', function () {
          client.end()
          done()
        })
      })
    })

    it('should support UTF-8 characters in payload', function (done) {
      var client = connect()

      client.once('connect', function () {
        client.publish('hello', '中国', function () {
          client.end()
          done()
        })
      })
    })

    it('should publish 10 QoS 2 and receive them', function (done) {
      var client = connect()
      var count = 0

      client.on('connect', function () {
        client.subscribe('test')
        client.publish('test', 'test', { qos: 2 })
      })

      client.on('message', function () {
        if (count >= 10) {
          client.end()
          done()
        } else {
          client.publish('test', 'test', { qos: 2 })
        }
      })

      server.once('client', function (serverClient) {
        serverClient.on('offline', function () {
          client.end()
          done('error went offline... didnt see this happen')
        })

        serverClient.on('subscribe', function () {
          serverClient.on('publish', function (packet) {
            serverClient.publish(packet)
          })
        })

        serverClient.on('pubrel', function () {
          count++
        })
      })
    })

    function testQosHandleMessage (qos, done) {
      var client = connect()

      var messageEventCount = 0
      var handleMessageCount = 0

      client.handleMessage = function (packet, callback) {
        setTimeout(function () {
          handleMessageCount++
          // next message event should not emit until handleMessage completes
          handleMessageCount.should.equal(messageEventCount)
          if (handleMessageCount === 10) {
            setTimeout(function () {
              client.end()
              done()
            })
          }
          callback()
        }, 100)
      }

      client.on('message', function (topic, message, packet) {
        messageEventCount++
      })

      client.on('connect', function () {
        client.subscribe('test')
      })

      server.once('client', function (serverClient) {
        serverClient.on('offline', function () {
          client.end()
          done('error went offline... didnt see this happen')
        })

        serverClient.on('subscribe', function () {
          for (var i = 0; i < 10; i++) {
            serverClient.publish({
              messageId: i,
              topic: 'test',
              payload: 'test' + i,
              qos: qos
            })
          }
        })
      })
    }

    it('should publish 10 QoS 0 and receive them only when `handleMessage` finishes', function (done) {
      testQosHandleMessage(0, done)
    })

    it('should publish 10 QoS 1 and receive them only when `handleMessage` finishes', function (done) {
      testQosHandleMessage(1, done)
    })

    it('should publish 10 QoS 2 and receive them only when `handleMessage` finishes', function (done) {
      testQosHandleMessage(2, done)
    })

    it('should not send a `puback` if the execution of `handleMessage` fails for messages with QoS `1`', function (done) {
      var client = connect()

      client.handleMessage = function (packet, callback) {
        callback(new Error('Error thrown by the application'))
      }

      client._sendPacket = sinon.spy()

      client._handlePublish({
        messageId: Math.floor(65535 * Math.random()),
        topic: 'test',
        payload: 'test',
        qos: 1
      }, function (err) {
        should.exist(err)
      })

      client._sendPacket.callCount.should.equal(0)
      client.end()
      client.on('connect', function () { done() })
    })

    it('should silently ignore errors thrown by `handleMessage` and return when no callback is passed ' +
      'into `handlePublish` method', function (done) {
      var client = connect()

      client.handleMessage = function (packet, callback) {
        callback(new Error('Error thrown by the application'))
      }

      try {
        client._handlePublish({
          messageId: Math.floor(65535 * Math.random()),
          topic: 'test',
          payload: 'test',
          qos: 1
        })
        done()
      } catch (err) {
        done(err)
      } finally {
        client.end()
      }
    })

    it('should handle error with async incoming store in QoS 2 `handlePublish` method', function (done) {
      function AsyncStore () {
        if (!(this instanceof AsyncStore)) {
          return new AsyncStore()
        }
      }
      AsyncStore.prototype.put = function (packet, cb) {
        process.nextTick(function () {
          cb(new Error('Error'))
        })
      }
      var store = new AsyncStore()
      var client = connect({incomingStore: store})

      client._handlePublish({
        messageId: 1,
        topic: 'test',
        payload: 'test',
        qos: 2
      }, function () {
        done()
        client.end()
      })
    })

    it('should handle error with async incoming store in QoS 2 `handlePubrel` method', function (done) {
      function AsyncStore () {
        if (!(this instanceof AsyncStore)) {
          return new AsyncStore()
        }
      }
      AsyncStore.prototype.put = function (packet, cb) {
        process.nextTick(function () {
          cb(new Error('Error'))
        })
      }
      AsyncStore.prototype.get = function (packet, cb) {
        process.nextTick(function () {
          cb(null, {cmd: 'publish'})
        })
      }
      var store = new AsyncStore()
      var client = connect({incomingStore: store})

      client._handlePubrel({
        messageId: 1,
        qos: 2
      }, function () {
        done()
        client.end()
      })
    })

    it('should handle success with async incoming store in QoS 2 `handlePubrel` method', function (done) {
      var putComplete = false
      function AsyncStore () {
        if (!(this instanceof AsyncStore)) {
          return new AsyncStore()
        }
      }
      AsyncStore.prototype.put = function (packet, cb) {
        process.nextTick(function () {
          putComplete = true
          cb(null)
        })
      }
      AsyncStore.prototype.get = function (packet, cb) {
        process.nextTick(function () {
          cb(null, {cmd: 'publish'})
        })
      }
      var store = new AsyncStore()
      var client = connect({incomingStore: store})

      client._handlePubrel({
        messageId: 1,
        qos: 2
      }, function () {
        putComplete.should.equal(true)
        done()
        client.end()
      })
    })

    it('should handle error with async incoming store in QoS 1 `handlePublish` method', function (done) {
      function AsyncStore () {
        if (!(this instanceof AsyncStore)) {
          return new AsyncStore()
        }
      }
      AsyncStore.prototype.put = function (packet, cb) {
        process.nextTick(function () {
          cb(null, 'Error')
        })
      }
      var store = new AsyncStore()
      var client = connect({incomingStore: store})

      client._handlePublish({
        messageId: 1,
        topic: 'test',
        payload: 'test',
        qos: 1
      }, function () {
        done()
        client.end()
      })
    })

    it('should not send a `pubcomp` if the execution of `handleMessage` fails for messages with QoS `2`', function (done) {
      var store = new Store()
      var client = connect({incomingStore: store})

      var messageId = Math.floor(65535 * Math.random())
      var topic = 'test'
      var payload = 'test'
      var qos = 2

      client.handleMessage = function (packet, callback) {
        callback(new Error('Error thrown by the application'))
      }

      client.once('connect', function () {
        client.subscribe(topic, {qos: 2})

        store.put({
          messageId: messageId,
          topic: topic,
          payload: payload,
          qos: qos,
          cmd: 'publish'
        }, function () {
          // cleans up the client
          client.end()

          client._sendPacket = sinon.spy()
          client._handlePubrel({cmd: 'pubrel', messageId: messageId}, function (err) {
            should.exist(err)
          })
          client._sendPacket.callCount.should.equal(0)
          done()
        })
      })
    })

    it('should silently ignore errors thrown by `handleMessage` and return when no callback is passed ' +
      'into `handlePubrel` method', function (done) {
      var store = new Store()
      var client = connect({incomingStore: store})

      var messageId = Math.floor(65535 * Math.random())
      var topic = 'test'
      var payload = 'test'
      var qos = 2

      client.handleMessage = function (packet, callback) {
        callback(new Error('Error thrown by the application'))
      }

      client.once('connect', function () {
        client.subscribe(topic, {qos: 2})

        store.put({
          messageId: messageId,
          topic: topic,
          payload: payload,
          qos: qos,
          cmd: 'publish'
        }, function () {
          try {
            client._handlePubrel({cmd: 'pubrel', messageId: messageId})
            done()
          } catch (err) {
            done(err)
          } finally {
            client.end()
          }
        })
      })
    })
  })

  describe('unsubscribing', function () {
    it('should send an unsubscribe packet (offline)', function (done) {
      var client = connect()

      client.unsubscribe('test')

      server.once('client', function (serverClient) {
        serverClient.once('unsubscribe', function (packet) {
          packet.unsubscriptions.should.containEql('test')
          client.end()
          done()
        })
      })
    })

    it('should send an unsubscribe packet', function (done) {
      var client = connect()
      var topic = 'topic'

      client.once('connect', function () {
        client.unsubscribe(topic)
      })

      server.once('client', function (serverClient) {
        serverClient.once('unsubscribe', function (packet) {
          packet.unsubscriptions.should.containEql(topic)
          client.end()
          done()
        })
      })
    })

    it('should emit a packetsend event', function (done) {
      var client = connect()
      var testTopic = 'testTopic'

      client.once('connect', function () {
        client.subscribe(testTopic)
      })

      client.on('packetsend', function (packet) {
        if (packet.cmd === 'subscribe') {
          client.end()
          done()
        }
      })
    })

    it('should emit a packetreceive event', function (done) {
      var client = connect()
      var testTopic = 'testTopic'

      client.once('connect', function () {
        client.subscribe(testTopic)
      })

      client.on('packetreceive', function (packet) {
        if (packet.cmd === 'suback') {
          client.end()
          done()
        }
      })
    })

    it('should accept an array of unsubs', function (done) {
      var client = connect()
      var topics = ['topic1', 'topic2']

      client.once('connect', function () {
        client.unsubscribe(topics)
      })

      server.once('client', function (serverClient) {
        serverClient.once('unsubscribe', function (packet) {
          packet.unsubscriptions.should.eql(topics)
          done()
        })
      })
    })

    it('should fire a callback on unsuback', function (done) {
      var client = connect()
      var topic = 'topic'

      client.once('connect', function () {
        client.unsubscribe(topic, done)
      })

      server.once('client', function (serverClient) {
        serverClient.once('unsubscribe', function (packet) {
          serverClient.unsuback(packet)
          client.end()
        })
      })
    })

    it('should unsubscribe from a chinese topic', function (done) {
      var client = connect()
      var topic = '中国'

      client.once('connect', function () {
        client.unsubscribe(topic)
      })

      server.once('client', function (serverClient) {
        serverClient.once('unsubscribe', function (packet) {
          packet.unsubscriptions.should.containEql(topic)
          client.end()
          done()
        })
      })
    })
  })

  describe('keepalive', function () {
    var clock

    beforeEach(function () {
      clock = sinon.useFakeTimers()
    })

    afterEach(function () {
      clock.restore()
    })

    it('should checkPing at keepalive interval', function (done) {
      var interval = 3
      var client = connect({ keepalive: interval })

      client._checkPing = sinon.spy()

      client.once('connect', function () {
        clock.tick(interval * 1000)
        client._checkPing.callCount.should.equal(1)

        clock.tick(interval * 1000)
        client._checkPing.callCount.should.equal(2)

        clock.tick(interval * 1000)
        client._checkPing.callCount.should.equal(3)

        client.end()
        done()
      })
    })

    it('should not checkPing if publishing at a higher rate than keepalive', function (done) {
      var intervalMs = 3000
      var client = connect({keepalive: intervalMs / 1000})

      client._checkPing = sinon.spy()

      client.once('connect', function () {
        client.publish('foo', 'bar')
        clock.tick(intervalMs - 1)
        client.publish('foo', 'bar')
        clock.tick(2)
        client._checkPing.callCount.should.equal(0)
        client.end()
        done()
      })
    })

    it('should checkPing if publishing at a higher rate than keepalive and reschedulePings===false', function (done) {
      var intervalMs = 3000
      var client = connect({
        keepalive: intervalMs / 1000,
        reschedulePings: false
      })

      client._checkPing = sinon.spy()

      client.once('connect', function () {
        client.publish('foo', 'bar')
        clock.tick(intervalMs - 1)
        client.publish('foo', 'bar')
        clock.tick(2)
        client._checkPing.callCount.should.equal(1)
        client.end()
        done()
      })
    })
  })

  describe('pinging', function () {
    it('should set a ping timer', function (done) {
      var client = connect({keepalive: 3})
      client.once('connect', function () {
        should.exist(client.pingTimer)
        client.end()
        done()
      })
    })

    it('should not set a ping timer keepalive=0', function (done) {
      var client = connect({keepalive: 0})
      client.on('connect', function () {
        should.not.exist(client.pingTimer)
        client.end()
        done()
      })
    })

    it('should reconnect if pingresp is not sent', function (done) {
      var client = connect({keepalive: 1, reconnectPeriod: 100})

      // Fake no pingresp being send by stubbing the _handlePingresp function
      client._handlePingresp = function () {}

      client.once('connect', function () {
        client.once('connect', function () {
          client.end()
          done()
        })
      })
    })

    it('should not reconnect if pingresp is successful', function (done) {
      var client = connect({keepalive: 100})
      client.once('close', function () {
        done(new Error('Client closed connection'))
      })
      setTimeout(done, 1000)
    })

    it('should defer the next ping when sending a control packet', function (done) {
      var client = connect({keepalive: 1})

      client.once('connect', function () {
        client._checkPing = sinon.spy()

        client.publish('foo', 'bar')
        setTimeout(function () {
          client._checkPing.callCount.should.equal(0)
          client.publish('foo', 'bar')

          setTimeout(function () {
            client._checkPing.callCount.should.equal(0)
            client.publish('foo', 'bar')

            setTimeout(function () {
              client._checkPing.callCount.should.equal(0)
              done()
            }, 75)
          }, 75)
        }, 75)
      })
    })
  })

  describe('subscribing', function () {
    it('should send a subscribe message (offline)', function (done) {
      var client = connect()

      client.subscribe('test')

      server.once('client', function (serverClient) {
        serverClient.once('subscribe', function () {
          done()
        })
      })
    })

    it('should send a subscribe message', function (done) {
      var client = connect()
      var topic = 'test'

      client.once('connect', function () {
        client.subscribe(topic)
      })

      server.once('client', function (serverClient) {
        serverClient.once('subscribe', function (packet) {
          packet.subscriptions.should.containEql({
            topic: topic,
            qos: 0
          })
          done()
        })
      })
    })

    it('should emit a packetsend event', function (done) {
      var client = connect()
      var testTopic = 'testTopic'

      client.once('connect', function () {
        client.subscribe(testTopic)
      })

      client.on('packetsend', function (packet) {
        if (packet.cmd === 'subscribe') {
          done()
        }
      })
    })

    it('should emit a packetreceive event', function (done) {
      var client = connect()
      var testTopic = 'testTopic'

      client.once('connect', function () {
        client.subscribe(testTopic)
      })

      client.on('packetreceive', function (packet) {
        if (packet.cmd === 'suback') {
          done()
        }
      })
    })

    it('should accept an array of subscriptions', function (done) {
      var client = connect()
      var subs = ['test1', 'test2']

      client.once('connect', function () {
        client.subscribe(subs)
      })

      server.once('client', function (serverClient) {
        serverClient.once('subscribe', function (packet) {
          // i.e. [{topic: 'a', qos: 0}, {topic: 'b', qos: 0}]
          var expected = subs.map(function (i) {
            return {topic: i, qos: 0}
          })

          packet.subscriptions.should.eql(expected)
          done()
        })
      })
    })

    it('should accept an hash of subscriptions', function (done) {
      var client = connect()
      var topics = {
        test1: 0,
        test2: 1
      }

      client.once('connect', function () {
        client.subscribe(topics)
      })

      server.once('client', function (serverClient) {
        serverClient.once('subscribe', function (packet) {
          var k
          var expected = []

          for (k in topics) {
            if (topics.hasOwnProperty(k)) {
              expected.push({
                topic: k,
                qos: topics[k]
              })
            }
          }

          packet.subscriptions.should.eql(expected)
          done()
        })
      })
    })

    it('should accept an options parameter', function (done) {
      var client = connect()
      var topic = 'test'
      var opts = {qos: 1}

      client.once('connect', function () {
        client.subscribe(topic, opts)
      })

      server.once('client', function (serverClient) {
        serverClient.once('subscribe', function (packet) {
          var expected = [{
            topic: topic,
            qos: 1
          }]

          packet.subscriptions.should.eql(expected)
          done()
        })
      })
    })

    it('should subscribe with the default options for an empty options parameter', function (done) {
      var client = connect()
      var topic = 'test'
      var defaultOpts = {qos: 0}

      client.once('connect', function () {
        client.subscribe(topic, {})
      })

      server.once('client', function (serverClient) {
        serverClient.once('subscribe', function (packet) {
          packet.subscriptions.should.containEql({
            topic: topic,
            qos: defaultOpts.qos
          })
          done()
        })
      })
    })

    it('should fire a callback on suback', function (done) {
      var client = connect()
      var topic = 'test'

      client.once('connect', function () {
        client.subscribe(topic, { qos: 2 }, function (err, granted) {
          if (err) {
            done(err)
          } else {
            should.exist(granted, 'granted not given')
            granted.should.containEql({topic: 'test', qos: 2})
            done()
          }
        })
      })
    })

    it('should fire a callback with error if disconnected (options provided)', function (done) {
      var client = connect()
      var topic = 'test'
      client.once('connect', function () {
        client.end(true, function () {
          client.subscribe(topic, {qos: 2}, function (err, granted) {
            should.not.exist(granted, 'granted given')
            should.exist(err, 'no error given')
            done()
          })
        })
      })
    })

    it('should fire a callback with error if disconnected (options not provided)', function (done) {
      var client = connect()
      var topic = 'test'

      client.once('connect', function () {
        client.end(true, function () {
          client.subscribe(topic, function (err, granted) {
            should.not.exist(granted, 'granted given')
            should.exist(err, 'no error given')
            done()
          })
        })
      })
    })

    it('should subscribe with a chinese topic', function (done) {
      var client = connect()
      var topic = '中国'

      client.once('connect', function () {
        client.subscribe(topic)
      })

      server.once('client', function (serverClient) {
        serverClient.once('subscribe', function (packet) {
          packet.subscriptions.should.containEql({
            topic: topic,
            qos: 0
          })
          done()
        })
      })
    })
  })

  describe('receiving messages', function () {
    it('should fire the message event', function (done) {
      var client = connect()
      var testPacket = {
        topic: 'test',
        payload: 'message',
        retain: true,
        qos: 1,
        messageId: 5
      }

      client.subscribe(testPacket.topic)
      client.once('message', function (topic, message, packet) {
        topic.should.equal(testPacket.topic)
        message.toString().should.equal(testPacket.payload)
        packet.should.equal(packet)
        client.end()
        done()
      })

      server.once('client', function (serverClient) {
        serverClient.on('subscribe', function () {
          serverClient.publish(testPacket)
        })
      })
    })

    it('should emit a packetreceive event', function (done) {
      var client = connect()
      var testPacket = {
        topic: 'test',
        payload: 'message',
        retain: true,
        qos: 1,
        messageId: 5
      }

      client.subscribe(testPacket.topic)
      client.on('packetreceive', function (packet) {
        if (packet.cmd === 'publish') {
          packet.qos.should.equal(1)
          packet.topic.should.equal(testPacket.topic)
          packet.payload.toString().should.equal(testPacket.payload)
          packet.retain.should.equal(true)
          client.end()
          done()
        }
      })

      server.once('client', function (serverClient) {
        serverClient.on('subscribe', function () {
          serverClient.publish(testPacket)
        })
      })
    })

    it('should support binary data', function (done) {
      var client = connect({ encoding: 'binary' })
      var testPacket = {
        topic: 'test',
        payload: 'message',
        retain: true,
        qos: 1,
        messageId: 5
      }

      client.subscribe(testPacket.topic)
      client.once('message', function (topic, message, packet) {
        topic.should.equal(testPacket.topic)
        message.should.be.an.instanceOf(Buffer)
        message.toString().should.equal(testPacket.payload)
        packet.should.equal(packet)
        done()
      })

      server.once('client', function (serverClient) {
        serverClient.on('subscribe', function () {
          serverClient.publish(testPacket)
        })
      })
    })

    it('should emit a message event (qos=2)', function (done) {
      var client = connect()
      var testPacket = {
        topic: 'test',
        payload: 'message',
        retain: true,
        qos: 2,
        messageId: 5
      }

      server.testPublish = testPacket

      client.subscribe(testPacket.topic)
      client.once('message', function (topic, message, packet) {
        topic.should.equal(testPacket.topic)
        message.toString().should.equal(testPacket.payload)
        packet.should.equal(packet)
        done()
      })

      server.once('client', function (serverClient) {
        serverClient.on('subscribe', function () {
          serverClient.publish(testPacket)
        })
      })
    })

    it('should emit a message event (qos=2) - repeated publish', function (done) {
      var client = connect()
      var testPacket = {
        topic: 'test',
        payload: 'message',
        retain: true,
        qos: 2,
        messageId: 5
      }

      server.testPublish = testPacket

      client.subscribe(testPacket.topic)
      client.on('message', function (topic, message, packet) {
        topic.should.equal(testPacket.topic)
        message.toString().should.equal(testPacket.payload)
        packet.should.equal(packet)
        done()
      })

      server.once('client', function (serverClient) {
        serverClient.on('subscribe', function () {
          serverClient.publish(testPacket)
          // twice, should be ignored
          serverClient.publish(testPacket)
        })
      })
    })

    it('should support chinese topic', function (done) {
      var client = connect({ encoding: 'binary' })
      var testPacket = {
        topic: '国',
        payload: 'message',
        retain: true,
        qos: 1,
        messageId: 5
      }

      client.subscribe(testPacket.topic)
      client.once('message', function (topic, message, packet) {
        topic.should.equal(testPacket.topic)
        message.should.be.an.instanceOf(Buffer)
        message.toString().should.equal(testPacket.payload)
        packet.should.equal(packet)
        done()
      })

      server.once('client', function (serverClient) {
        serverClient.on('subscribe', function () {
          serverClient.publish(testPacket)
        })
      })
    })
  })

  describe('qos handling', function () {
    it('should follow qos 0 semantics (trivial)', function (done) {
      var client = connect()
      var testTopic = 'test'
      var testMessage = 'message'

      client.once('connect', function () {
        client.subscribe(testTopic, {qos: 0})
      })

      server.once('client', function (serverClient) {
        serverClient.once('subscribe', function () {
          serverClient.publish({
            topic: testTopic,
            payload: testMessage,
            qos: 0,
            retain: false
          })
          done()
        })
      })
    })

    it('should follow qos 1 semantics', function (done) {
      var client = connect()
      var testTopic = 'test'
      var testMessage = 'message'
      var mid = 50

      client.once('connect', function () {
        client.subscribe(testTopic, {qos: 1})
      })

      server.once('client', function (serverClient) {
        serverClient.once('subscribe', function () {
          serverClient.publish({
            topic: testTopic,
            payload: testMessage,
            messageId: mid,
            qos: 1
          })
        })

        serverClient.once('puback', function (packet) {
          packet.messageId.should.equal(mid)
          done()
        })
      })
    })

    it('should follow qos 2 semantics', function (done) {
      var client = connect()
      var testTopic = 'test'
      var testMessage = 'message'
      var mid = 253

      client.once('connect', function () {
        client.subscribe(testTopic, {qos: 2})
      })

      server.once('client', function (serverClient) {
        serverClient.once('subscribe', function () {
          serverClient.publish({
            topic: testTopic,
            payload: testMessage,
            qos: 2,
            messageId: mid
          })
        })

        serverClient.once('pubcomp', function () {
          done()
        })
      })
    })
  })

  describe('auto reconnect', function () {
    it('should mark the client disconnecting if #end called', function () {
      var client = connect()

      client.end()
      client.disconnecting.should.eql(true)
    })

    it('should reconnect after stream disconnect', function (done) {
      var client = connect()
      var tryReconnect = true

      client.on('connect', function () {
        if (tryReconnect) {
          client.stream.end()
          tryReconnect = false
        } else {
          client.end()
          done()
        }
      })
    })

    it('should emit \'reconnect\' when reconnecting', function (done) {
      var client = connect()
      var tryReconnect = true
      var reconnectEvent = false

      client.on('reconnect', function () {
        reconnectEvent = true
      })

      client.on('connect', function () {
        if (tryReconnect) {
          client.stream.end()
          tryReconnect = false
        } else {
          reconnectEvent.should.equal(true)
          client.end()
          done()
        }
      })
    })

    it('should emit \'offline\' after going offline', function (done) {
      var client = connect()
      var tryReconnect = true
      var offlineEvent = false

      client.on('offline', function () {
        offlineEvent = true
      })

      client.on('connect', function () {
        if (tryReconnect) {
          client.stream.end()
          tryReconnect = false
        } else {
          offlineEvent.should.equal(true)
          client.end()
          done()
        }
      })
    })

    it('should not reconnect if it was ended by the user', function (done) {
      var client = connect()

      client.on('connect', function () {
        client.end()
        done() // it will raise an exception if called two times
      })
    })

    it('should setup a reconnect timer on disconnect', function (done) {
      var client = connect()

      client.once('connect', function () {
        should.not.exist(client.reconnectTimer)
        client.stream.end()
      })

      client.once('close', function () {
        should.exist(client.reconnectTimer)
        client.end()
        done()
      })
    })

    it('should allow specification of a reconnect period', function (done) {
      var end
      var period = 200
      var client = connect({reconnectPeriod: period})
      var reconnect = false
      var start = Date.now()

      client.on('connect', function () {
        if (!reconnect) {
          client.stream.end()
          reconnect = true
        } else {
          client.end()
          end = Date.now()
          if (end - start >= period) {
            // Connected in about 2 seconds, that's good enough
            done()
          } else {
            done(new Error('Strange reconnect period'))
          }
        }
      })
    })

    it('should always cleanup successfully on reconnection', function (done) {
      var client = connect({host: 'this_hostname_should_not_exist', connectTimeout: 0, reconnectPeriod: 1})
      setTimeout(client.end.bind(client, done), 50)
    })

    it('should resend in-flight QoS 1 publish messages from the client', function (done) {
      var client = connect({reconnectPeriod: 200})
      var serverPublished = false
      var clientCalledBack = false

      server.once('client', function (serverClient) {
        serverClient.on('connect', function () {
          setImmediate(function () {
            serverClient.stream.destroy()
          })
        })

        server.once('client', function (serverClientNew) {
          serverClientNew.on('publish', function () {
            serverPublished = true
            check()
          })
        })
      })

      client.publish('hello', 'world', { qos: 1 }, function () {
        clientCalledBack = true
        check()
      })

      function check () {
        if (serverPublished && clientCalledBack) {
          client.end()
          done()
        }
      }
    })

    it('should not resend in-flight publish messages if disconnecting', function (done) {
      var client = connect({reconnectPeriod: 200})
      var serverPublished = false
      var clientCalledBack = false
      server.once('client', function (serverClient) {
        serverClient.on('connect', function () {
          setImmediate(function () {
            serverClient.stream.destroy()
            client.end()
            serverPublished.should.be.false()
            clientCalledBack.should.be.false()
            done()
          })
        })
        server.once('client', function (serverClientNew) {
          serverClientNew.on('publish', function () {
            serverPublished = true
          })
        })
      })
      client.publish('hello', 'world', { qos: 1 }, function () {
        clientCalledBack = true
      })
    })

    it('should resend in-flight QoS 2 publish messages from the client', function (done) {
      var client = connect({reconnectPeriod: 200})
      var serverPublished = false
      var clientCalledBack = false

      server.once('client', function (serverClient) {
        // ignore errors
        serverClient.on('error', function () {})
        serverClient.on('publish', function () {
          setImmediate(function () {
            serverClient.stream.destroy()
          })
        })

        server.once('client', function (serverClientNew) {
          serverClientNew.on('pubrel', function () {
            serverPublished = true
            check()
          })
        })
      })

      client.publish('hello', 'world', { qos: 2 }, function () {
        clientCalledBack = true
        check()
      })

      function check () {
        if (serverPublished && clientCalledBack) {
          client.end()
          done()
        }
      }
    })

    it('should not resend in-flight QoS 1 removed publish messages from the client', function (done) {
      var client = connect({reconnectPeriod: 200})
      var clientCalledBack = false

      server.once('client', function (serverClient) {
        serverClient.on('connect', function () {
          setImmediate(function () {
            serverClient.stream.destroy()
          })
        })

        server.once('client', function (serverClientNew) {
          serverClientNew.on('publish', function () {
            should.fail()
            done()
          })
        })
      })

      client.publish('hello', 'world', { qos: 1 }, function (err) {
        clientCalledBack = true
        should(err.message).be.equal('Message removed')
      })
      should(Object.keys(client.outgoing).length).be.equal(1)
      should(client.outgoingStore._inflights.size).be.equal(1)
      client.removeOutgoingMessage(client.getLastMessageId())
      should(Object.keys(client.outgoing).length).be.equal(0)
      should(client.outgoingStore._inflights.size).be.equal(0)
      clientCalledBack.should.be.true()
      client.end()
      done()
    })

    it('should not resend in-flight QoS 2 removed publish messages from the client', function (done) {
      var client = connect({reconnectPeriod: 200})
      var clientCalledBack = false

      server.once('client', function (serverClient) {
        serverClient.on('connect', function () {
          setImmediate(function () {
            serverClient.stream.destroy()
          })
        })

        server.once('client', function (serverClientNew) {
          serverClientNew.on('publish', function () {
            should.fail()
            done()
          })
        })
      })

      client.publish('hello', 'world', { qos: 2 }, function (err) {
        clientCalledBack = true
        should(err.message).be.equal('Message removed')
      })
      should(Object.keys(client.outgoing).length).be.equal(1)
      should(client.outgoingStore._inflights.size).be.equal(1)
      client.removeOutgoingMessage(client.getLastMessageId())
      should(Object.keys(client.outgoing).length).be.equal(0)
      should(client.outgoingStore._inflights.size).be.equal(0)
      clientCalledBack.should.be.true()
      client.end()
      done()
    })

    it('should resubscribe when reconnecting', function (done) {
      var client = connect({ reconnectPeriod: 100 })
      var tryReconnect = true
      var reconnectEvent = false

      client.on('reconnect', function () {
        reconnectEvent = true
      })

      client.on('connect', function () {
        if (tryReconnect) {
          client.subscribe('hello', function () {
            client.stream.end()

            server.once('client', function (serverClient) {
              serverClient.on('subscribe', function () {
                client.end()
                done()
              })
            })
          })

          tryReconnect = false
        } else {
          reconnectEvent.should.equal(true)
        }
      })
    })

    it('should not resubscribe when reconnecting if resubscribe is disabled', function (done) {
      var client = connect({ reconnectPeriod: 100, resubscribe: false })
      var tryReconnect = true
      var reconnectEvent = false

      client.on('reconnect', function () {
        reconnectEvent = true
      })

      client.on('connect', function () {
        if (tryReconnect) {
          client.subscribe('hello', function () {
            client.stream.end()

            server.once('client', function (serverClient) {
              serverClient.on('subscribe', function () {
                should.fail()
              })
            })
          })

          tryReconnect = false
        } else {
          reconnectEvent.should.equal(true)
          should(Object.keys(client._resubscribeTopics).length).be.equal(0)
          done()
        }
      })
    })

    it('should not resubscribe when reconnecting if suback is error', function (done) {
      var tryReconnect = true
      var reconnectEvent = false
      var server2 = new Server(function (c) {
        c.on('connect', function (packet) {
          c.connack({returnCode: 0})
        })
        c.on('subscribe', function (packet) {
          c.suback({
            messageId: packet.messageId,
            granted: packet.subscriptions.map(function (e) {
              return e.qos | 0x80
            })
          })
          c.pubrel({ messageId: Math.floor(Math.random() * 9000) + 1000 })
        })
      })

      server2.listen(port + 49, function () {
        var client = mqtt.connect({
          port: port + 49,
          host: 'localhost',
          reconnectPeriod: 100
        })

        client.on('reconnect', function () {
          reconnectEvent = true
        })

        client.on('connect', function () {
          if (tryReconnect) {
            client.subscribe('hello', function () {
              client.stream.end()

              server.once('client', function (serverClient) {
                serverClient.on('subscribe', function () {
                  should.fail()
                })
              })
            })
            tryReconnect = false
          } else {
            reconnectEvent.should.equal(true)
            should(Object.keys(client._resubscribeTopics).length).be.equal(0)
            server2.close()
            done()
          }
        })
      })
    })

    it('should preserved incomingStore after disconnecting if clean is false', function (done) {
      var reconnect = false
      var client = {}
      var incomingStore = new mqtt.Store({ clean: false })
      var outgoingStore = new mqtt.Store({ clean: false })
      var server2 = new Server(function (c) {
        c.on('connect', function (packet) {
          c.connack({returnCode: 0})
          if (reconnect) {
            c.pubrel({ messageId: 1 })
          }
        })
        c.on('subscribe', function (packet) {
          c.suback({
            messageId: packet.messageId,
            granted: packet.subscriptions.map(function (e) {
              return e.qos
            })
          })
          c.publish({ topic: 'topic', payload: 'payload', qos: 2, messageId: 1, retain: false })
        })
        c.on('pubrec', function (packet) {
          client.end(false, function () {
            client.reconnect({
              incomingStore: incomingStore,
              outgoingStore: outgoingStore
            })
          })
        })
        c.on('pubcomp', function (packet) {
          client.end()
          server2.close()
          done()
        })
      })

      server2.listen(port + 50, function () {
        client = mqtt.connect({
          port: port + 50,
          host: 'localhost',
          clean: false,
          clientId: 'cid1',
          reconnectPeriod: 0,
          incomingStore: incomingStore,
          outgoingStore: outgoingStore
        })

        client.on('connect', function () {
          if (!reconnect) {
            client.subscribe('test', {qos: 2}, function () {
            })
            reconnect = true
          }
        })
        client.on('message', function (topic, message) {
          topic.should.equal('topic')
          message.toString().should.equal('payload')
        })
      })
    })

    it('should resend in-flight QoS 1 publish messages from the client if clean is false', function (done) {
      var reconnect = false
      var client = {}
      var incomingStore = new mqtt.Store({ clean: false })
      var outgoingStore = new mqtt.Store({ clean: false })
      var server2 = new Server(function (c) {
        c.on('connect', function (packet) {
          c.connack({returnCode: 0})
        })
        c.on('publish', function (packet) {
          if (reconnect) {
            server2.close()
            done()
          } else {
            client.end(true, function () {
              client.reconnect({
                incomingStore: incomingStore,
                outgoingStore: outgoingStore
              })
              reconnect = true
            })
          }
        })
      })

      server2.listen(port + 50, function () {
        client = mqtt.connect({
          port: port + 50,
          host: 'localhost',
          clean: false,
          clientId: 'cid1',
          reconnectPeriod: 0,
          incomingStore: incomingStore,
          outgoingStore: outgoingStore
        })

        client.on('connect', function () {
          if (!reconnect) {
            client.publish('topic', 'payload', {qos: 1})
          }
        })
        client.on('error', function () {})
      })
    })

    it('should resend in-flight QoS 2 publish messages from the client if clean is false', function (done) {
      var reconnect = false
      var client = {}
      var incomingStore = new mqtt.Store({ clean: false })
      var outgoingStore = new mqtt.Store({ clean: false })
      var server2 = new Server(function (c) {
        c.on('connect', function (packet) {
          c.connack({returnCode: 0})
        })
        c.on('publish', function (packet) {
          if (reconnect) {
            server2.close()
            done()
          } else {
            client.end(true, function () {
              client.reconnect({
                incomingStore: incomingStore,
                outgoingStore: outgoingStore
              })
              reconnect = true
            })
          }
        })
      })

      server2.listen(port + 50, function () {
        client = mqtt.connect({
          port: port + 50,
          host: 'localhost',
          clean: false,
          clientId: 'cid1',
          reconnectPeriod: 0,
          incomingStore: incomingStore,
          outgoingStore: outgoingStore
        })

        client.on('connect', function () {
          if (!reconnect) {
            client.publish('topic', 'payload', {qos: 2})
          }
        })
        client.on('error', function () {})
      })
    })

    it('should resend in-flight QoS 2 pubrel messages from the client if clean is false', function (done) {
      var reconnect = false
      var client = {}
      var incomingStore = new mqtt.Store({ clean: false })
      var outgoingStore = new mqtt.Store({ clean: false })
      var server2 = new Server(function (c) {
        c.on('connect', function (packet) {
          c.connack({returnCode: 0})
        })
        c.on('publish', function (packet) {
          if (!reconnect) {
            c.pubrec({messageId: packet.messageId})
          }
        })
        c.on('pubrel', function () {
          if (reconnect) {
            server2.close()
            done()
          } else {
            client.end(true, function () {
              client.reconnect({
                incomingStore: incomingStore,
                outgoingStore: outgoingStore
              })
              reconnect = true
            })
          }
        })
      })

      server2.listen(port + 50, function () {
        client = mqtt.connect({
          port: port + 50,
          host: 'localhost',
          clean: false,
          clientId: 'cid1',
          reconnectPeriod: 0,
          incomingStore: incomingStore,
          outgoingStore: outgoingStore
        })

        client.on('connect', function () {
          if (!reconnect) {
            client.publish('topic', 'payload', {qos: 2})
          }
        })
        client.on('error', function () {})
      })
    })

    it('should resend in-flight publish messages by published order', function (done) {
      var publishCount = 0
      var reconnect = false
      var disconnectOnce = true
      var client = {}
      var incomingStore = new mqtt.Store({ clean: false })
      var outgoingStore = new mqtt.Store({ clean: false })
      var server2 = new Server(function (c) {
        // errors are not interesting for this test
        // but they might happen on some platforms
        c.on('error', function () {})

        c.on('connect', function (packet) {
          c.connack({returnCode: 0})
        })
        c.on('publish', function (packet) {
          c.puback({messageId: packet.messageId})
          if (reconnect) {
            switch (publishCount++) {
              case 0:
                packet.payload.toString().should.equal('payload1')
                break
              case 1:
                packet.payload.toString().should.equal('payload2')
                break
              case 2:
                packet.payload.toString().should.equal('payload3')
                server2.close()
                done()
                break
            }
          } else {
            if (disconnectOnce) {
              client.end(true, function () {
                reconnect = true
                client.reconnect({
                  incomingStore: incomingStore,
                  outgoingStore: outgoingStore
                })
              })
              disconnectOnce = false
            }
          }
        })
      })

      server2.listen(port + 50, function () {
        client = mqtt.connect({
          port: port + 50,
          host: 'localhost',
          clean: false,
          clientId: 'cid1',
          reconnectPeriod: 0,
          incomingStore: incomingStore,
          outgoingStore: outgoingStore
        })

        client.nextId = 65535

        client.on('connect', function () {
          if (!reconnect) {
            client.publish('topic', 'payload1', {qos: 1})
            client.publish('topic', 'payload2', {qos: 1})
            client.publish('topic', 'payload3', {qos: 1})
          }
        })
        client.on('error', function () {})
      })
    })

    it('should be able to pub/sub if reconnect() is called at close handler', function (done) {
      var client = connect({ reconnectPeriod: 0 })
      var tryReconnect = true
      var reconnectEvent = false

      client.on('close', function () {
        if (tryReconnect) {
          tryReconnect = false
          client.reconnect()
        } else {
          reconnectEvent.should.equal(true)
          done()
        }
      })

      client.on('reconnect', function () {
        reconnectEvent = true
      })

      client.on('connect', function () {
        if (tryReconnect) {
          client.end()
        } else {
          client.subscribe('hello', function () {
            client.end()
          })
        }
      })
    })

    it('should be able to pub/sub if reconnect() is called at out of close handler', function (done) {
      var client = connect({ reconnectPeriod: 0 })
      var tryReconnect = true
      var reconnectEvent = false

      client.on('close', function () {
        if (tryReconnect) {
          tryReconnect = false
          setTimeout(function () {
            client.reconnect()
          }, 100)
        } else {
          reconnectEvent.should.equal(true)
          done()
        }
      })

      client.on('reconnect', function () {
        reconnectEvent = true
      })

      client.on('connect', function () {
        if (tryReconnect) {
          client.end()
        } else {
          client.subscribe('hello', function () {
            client.end()
          })
        }
      })
    })

    context('with alternate server client', function () {
      var cachedClientListeners

      beforeEach(function () {
        cachedClientListeners = server.listeners('client')
        server.removeAllListeners('client')
      })

      afterEach(function () {
        server.removeAllListeners('client')
        cachedClientListeners.forEach(function (listener) {
          server.on('client', listener)
        })
      })

      it('should resubscribe even if disconnect is before suback', function (done) {
        var client = mqtt.connect(Object.assign({ reconnectPeriod: 100 }, config))
        var subscribeCount = 0
        var connectCount = 0

        server.on('client', function (serverClient) {
          serverClient.on('connect', function () {
            connectCount++
            serverClient.connack({returnCode: 0})
          })

          serverClient.on('subscribe', function () {
            subscribeCount++

            // disconnect before sending the suback on the first subscribe
            if (subscribeCount === 1) {
              client.stream.end()
            }

            // after the second connection, confirm that the only two
            // subscribes have taken place, then cleanup and exit
            if (connectCount >= 2) {
              subscribeCount.should.equal(2)
              client.end(true, done)
            }
          })
        })

        client.subscribe('hello')
      })

      it('should resubscribe exactly once', function (done) {
        var client = mqtt.connect(Object.assign({ reconnectPeriod: 100 }, config))
        var subscribeCount = 0

        server.on('client', function (serverClient) {
          serverClient.on('connect', function () {
            serverClient.connack({returnCode: 0})
          })

          serverClient.on('subscribe', function () {
            subscribeCount++

            // disconnect before sending the suback on the first subscribe
            if (subscribeCount === 1) {
              client.stream.end()
            }

            // after the second connection, only two subs
            // subscribes have taken place, then cleanup and exit
            if (subscribeCount === 2) {
              client.end(true, done)
            }
          })
        })

        client.subscribe('hello')
      })
    })
  })
}