function emptyQueue (arg, done) {
  const client = this.client
  const persistence = client.broker.persistence
  const outgoing = persistence.outgoingStream(client)

  client.resume()

  pipeline(
    outgoing,
    through(function clearQueue (data, enc, next) {
      const packet = new QoSPacket(data, client)
      // Here we are deliberatly passing only the error
      // This is because there is no destination stream so the "client"
      // Object filled the buffer up to the highWaterMark preventing stored messages
      // being sent
      packet.writeCallback = (error, _client) => next(error)
      persistence.outgoingUpdate(client, packet, emptyQueueFilter)
    }),
    done
  )
}
// when QoS > 0 and session is true
when QoS > 0 and session is true
when QoS > 0 and session is true
function emptyQueue (arg, done) {
  const client = this.client
  const persistence = client.broker.persistence
  const outgoing = persistence.outgoingStream(client)
  client.resume()
  pipeline(
    outgoing,
    through(function clearQueue (data, enc, next) {
      const packet = new QoSPacket(data, client)
      packet.writeCallback = (error, _client) => next(error)
      persistence.outgoingUpdate(client, packet, emptyQueueFilter)
    }),
    done
  )
}
function emptyQueue ( arg , done ) { const client = this . client const persistence = client . broker . persistence const outgoing = persistence . outgoingStream ( client ) client . resume ( ) pipeline ( outgoing , through ( function clearQueue ( data , enc , next ) { const packet = new QoSPacket ( data , client ) packet . writeCallback = ( error , _client ) => next ( error ) persistence . outgoingUpdate ( client , packet , emptyQueueFilter ) } ) , done ) }
when QoS > 0 and session is true
when QoS > 0 and session is true
// Here we are deliberatly passing only the error // This is because there is no destination stream so the "client" // Object filled the buffer up to the highWaterMark preventing stored messages // being sent