function MqttClient (streamBuilder, options) {
  var k
  var that = this

  if (!(this instanceof MqttClient)) {
    return new MqttClient(streamBuilder, options)
  }

  this.options = options || {}

  // Defaults
  for (k in defaultConnectOptions) {
    if (typeof this.options[k] === 'undefined') {
      this.options[k] = defaultConnectOptions[k]
    } else {
      this.options[k] = options[k]
    }
  }

  this.options.clientId = this.options.clientId || defaultId()

  this.streamBuilder = streamBuilder

  // Inflight message storages
  this.outgoingStore = this.options.outgoingStore || new Store()
  this.incomingStore = this.options.incomingStore || new Store()

  // Should QoS zero messages be queued when the connection is broken?
  this.queueQoSZero = this.options.queueQoSZero === undefined ? true : this.options.queueQoSZero

  // map of subscribed topics to support reconnection
  this._subscribedTopics = {}

  // Ping timer, setup in _setupPingTimer
  this.pingTimer = null
  // Is the client connected?
  this.connected = false
  // Are we disconnecting?
  this.disconnecting = false
  // Packet queue
  this.queue = []
  // connack timer
  this.connackTimer = null
  // Reconnect timer
  this.reconnectTimer = null
  // MessageIDs starting with 1
  this.nextId = Math.floor(Math.random() * 65535)

  // Inflight callbacks
  this.outgoing = {}

  // Mark connected on connect
  this.on('connect', function () {
    if (this.disconnected) {
      return
    }

    this.connected = true
    var outStore = null
    outStore = this.outgoingStore.createStream()

    // Control of stored messages
    outStore.once('readable', function () {
      function storeDeliver () {
        var packet = outStore.read(1)
        var cb

        if (!packet) {
          return
        }

        // Avoid unnecesary stream read operations when disconnected
        if (!that.disconnecting && !that.reconnectTimer && that.options.reconnectPeriod > 0) {
          outStore.read(0)
          cb = that.outgoing[packet.messageId]
          that.outgoing[packet.messageId] = function (err, status) {
            // Ensure that the original callback passed in to publish gets invoked
            if (cb) {
              cb(err, status)
            }

            storeDeliver()
          }
          that._sendPacket(packet)
        } else if (outStore.destroy) {
          outStore.destroy()
        }
      }
      storeDeliver()
    })
    .on('error', this.emit.bind(this, 'error'))
  })

  // Mark disconnected on stream close
  this.on('close', function () {
    this.connected = false
    clearTimeout(this.connackTimer)
  })

  // Setup ping timer
  this.on('connect', this._setupPingTimer)

  // Send queued packets
  this.on('connect', function () {
    var queue = this.queue

    function deliver () {
      var entry = queue.shift()
      var packet = null

      if (!entry) {
        return
      }

      packet = entry.packet

      that._sendPacket(
        packet,
        function (err) {
          if (entry.cb) {
            entry.cb(err)
          }
          deliver()
        }
      )
    }

    deliver()
  })

  // resubscribe
  this.on('connect', function () {
    if (this.options.clean && Object.keys(this._subscribedTopics).length > 0) {
      this.subscribe(this._subscribedTopics)
    }
  })

  // Clear ping timer
  this.on('close', function () {
    if (that.pingTimer !== null) {
      that.pingTimer.clear()
      that.pingTimer = null
    }
  })

  // Setup reconnect timer on disconnect
  this.on('close', this._setupReconnect)

  events.EventEmitter.call(this)

  this._setupStream()
}
MqttClient constructor

@param {Stream} stream - stream
@param {Object} [options] - connection options
(see Connection#connect)