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

const { Transform } = require('readable-stream')

function validateTopic (topic, message) {
  const end = topic.length - 1
  const endMinus = end - 1
  const slashInPreEnd = endMinus > 0 && topic.charCodeAt(endMinus) !== 47
  if (topic.length === 0) { // [MQTT-3.8.3-3]
    return new Error('impossible to ' + message + ' to an empty topic')
  }
  for (let i = 0; i < topic.length; i++) {
    switch (topic.charCodeAt(i)) {
      case 35: { // #
        const notAtTheEnd = i !== end
        if (notAtTheEnd || slashInPreEnd) {
          return new Error('# is only allowed in ' + message + ' in the last position')
        }
        break
      }
      case 43: { // +
        const pastChar = i < end - 1 && topic.charCodeAt(i + 1) !== 47
        const preChar = i > 1 && topic.charCodeAt(i - 1) !== 47
        if (pastChar || preChar) {
          return new Error('+ is only allowed in ' + message + ' between /')
        }
        break
      }
    }
  }
}

function through (transform) {
  return new Transform({
    objectMode: true,
    transform
  })
}

module.exports = {
  validateTopic,
  through,
  $SYS_PREFIX: '$SYS/'
}


<|start_filename|>lib/qos-packet.js<|end_filename|>
'use strict'

const Packet = require('aedes-packet')
const util = require('util')

function QoSPacket (original, client) {
  Packet.call(this, original, client.broker)

  this.writeCallback = client._onError.bind(client)

  if (!original.messageId) {
    this.messageId = client._nextId
    if (client._nextId >= 65535) {
      client._nextId = 1
    } else {
      client._nextId++
    }
  } else {
    this.messageId = original.messageId
  }
}

util.inherits(QoSPacket, Packet)

module.exports = QoSPacket


<|start_filename|>example.js<|end_filename|>
'use strict'

const aedes = require('./aedes')()
const server = require('net').createServer(aedes.handle)
const httpServer = require('http').createServer()
const ws = require('websocket-stream')
const port = 1883
const wsPort = 8888

server.listen(port, function () {
  console.log('server listening on port', port)
})

ws.createServer({
  server: httpServer
}, aedes.handle)

httpServer.listen(wsPort, function () {
  console.log('websocket server listening on port', wsPort)
})

aedes.on('clientError', function (client, err) {
  console.log('client error', client.id, err.message, err.stack)
})

aedes.on('connectionError', function (client, err) {
  console.log('client error', client, err.message, err.stack)
})

aedes.on('publish', function (packet, client) {
  if (client) {
    console.log('message from client', client.id)
  }
})

aedes.on('subscribe', function (subscriptions, client) {
  if (client) {
    console.log('subscribe from client', subscriptions, client.id)
  }
})

aedes.on('client', function (client) {
  console.log('new client', client.id)
})


<|start_filename|>test/regr-21.js<|end_filename|>
'use strict'

const { test } = require('tap')
const { setup, connect } = require('./helper')

test('after an error, outstanding packets are discarded', function (t) {
  t.plan(1)

  const s = connect(setup(), {
    keepalive: 1000
  })
  t.teardown(s.broker.close.bind(s.broker))

  const packet = {
    cmd: 'publish',
    topic: 'hello',
    payload: 'world'
  }

  s.broker.mq.on('hello', function (msg, cb) {
    t.pass('first msg received')
    s.inStream.destroy(new Error('something went wrong'))
    cb()
    setImmediate(() => {
      packet.topic = 'foo'
      s.inStream.write(packet)
      s.inStream.write(packet)
    })
  })
  s.broker.mq.on('foo', function (msg, cb) {
    t.fail('msg received')
  })
  s.inStream.write(packet)
})


<|start_filename|>examples/clusters/index.js<|end_filename|>
const cluster = require('cluster')
const mqemitter = require('mqemitter-mongodb')
const mongoPersistence = require('aedes-persistence-mongodb')

const MONGO_URL = 'mongodb://127.0.0.1/aedes-clusters'

function startAedes () {
  const port = 1883

  const aedes = require('aedes')({
    id: 'BROKER_' + cluster.worker.id,
    mq: mqemitter({
      url: MONGO_URL
    }),
    persistence: mongoPersistence({
      url: MONGO_URL,
      // Optional ttl settings
      ttl: {
        packets: 300, // Number of seconds
        subscriptions: 300
      }
    })
  })

  const server = require('net').createServer(aedes.handle)

  server.listen(port, function () {
    console.log('Aedes listening on port:', port)
    aedes.publish({ topic: 'aedes/hello', payload: "I'm broker " + aedes.id })
  })

  aedes.on('subscribe', function (subscriptions, client) {
    console.log('MQTT client \x1b[32m' + (client ? client.id : client) +
            '\x1b[0m subscribed to topics: ' + subscriptions.map(s => s.topic).join('\n'), 'from broker', aedes.id)
  })

  aedes.on('unsubscribe', function (subscriptions, client) {
    console.log('MQTT client \x1b[32m' + (client ? client.id : client) +
            '\x1b[0m unsubscribed to topics: ' + subscriptions.join('\n'), 'from broker', aedes.id)
  })

  // fired when a client connects
  aedes.on('client', function (client) {
    console.log('Client Connected: \x1b[33m' + (client ? client.id : client) + '\x1b[0m', 'to broker', aedes.id)
  })

  // fired when a client disconnects
  aedes.on('clientDisconnect', function (client) {
    console.log('Client Disconnected: \x1b[31m' + (client ? client.id : client) + '\x1b[0m', 'to broker', aedes.id)
  })

  // fired when a message is published
  aedes.on('publish', async function (packet, client) {
    console.log('Client \x1b[31m' + (client ? client.id : 'BROKER_' + aedes.id) + '\x1b[0m has published', packet.payload.toString(), 'on', packet.topic, 'to broker', aedes.id)
  })
}

if (cluster.isMaster) {
  const numWorkers = require('os').cpus().length
  for (let i = 0; i < numWorkers; i++) {
    cluster.fork()
  }

  cluster.on('online', function (worker) {
    console.log('Worker ' + worker.process.pid + ' is online')
  })

  cluster.on('exit', function (worker, code, signal) {
    console.log('Worker ' + worker.process.pid + ' died with code: ' + code + ', and signal: ' + signal)
    console.log('Starting a new worker')
    cluster.fork()
  })
} else {
  startAedes()
}


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

const write = require('../write')

function PubAck (packet) {
  this.cmd = 'puback'
  this.messageId = packet.messageId
}

function PubRec (packet) {
  this.cmd = 'pubrec'
  this.messageId = packet.messageId
}

const publishActions = [
  authorizePublish,
  enqueuePublish
]
function handlePublish (client, packet, done) {
  const topic = packet.topic
  let err
  if (topic.length === 0) {
    err = new Error('empty topic not allowed in PUBLISH')
    return done(err)
  }
  if (topic.indexOf('#') > -1) {
    err = new Error('# is not allowed in PUBLISH')
    return done(err)
  }
  if (topic.indexOf('+') > -1) {
    err = new Error('+ is not allowed in PUBLISH')
    return done(err)
  }
  client.broker._series(client, publishActions, packet, done)
}

function enqueuePublish (packet, done) {
  const client = this

  switch (packet.qos) {
    case 2:
      client.broker.persistence.incomingStorePacket(client, packet, function (err) {
        if (err) { return done(err) }
        write(client, new PubRec(packet), done)
      })
      break
    case 1:
      write(client, new PubAck(packet), function (err) {
        if (err) { return done(err) }
        client.broker.publish(packet, client, done)
      })
      break
    case 0:
      client.broker.publish(packet, client, done)
      break
    default:
      // nothing to do
  }
}

function authorizePublish (packet, done) {
  this.broker.authorizePublish(this, packet, done)
}

module.exports = handlePublish


<|start_filename|>test/close_socket_by_other_party.js<|end_filename|>
'use strict'

const { test } = require('tap')
const EventEmitter = require('events')
const { setup, connect, subscribe } = require('./helper')
const aedes = require('../')

test('aedes is closed before client authenticate returns', function (t) {
  t.plan(1)

  const evt = new EventEmitter()
  const broker = aedes({
    authenticate: (client, username, password, done) => {
      evt.emit('AuthenticateBegin', client)
      setTimeout(function () {
        done(null, true)
      }, 2000)
    }
  })

  broker.on('client', function (client) {
    t.fail('should no client registration')
  })
  broker.on('connackSent', function () {
    t.fail('should no connack be sent')
  })
  broker.on('clientError', function (client, err) {
    t.error(err)
  })

  connect(setup(broker))

  evt.on('AuthenticateBegin', function (client) {
    t.equal(broker.connectedClients, 0)
    broker.close()
  })
})

test('client is closed before authenticate returns', function (t) {
  t.plan(1)

  const evt = new EventEmitter()
  const broker = aedes({
    authenticate: async (client, username, password, done) => {
      evt.emit('AuthenticateBegin', client)
      setTimeout(function () {
        done(null, true)
      }, 2000)
    }
  })
  t.teardown(broker.close.bind(broker))

  broker.on('client', function (client) {
    t.fail('should no client registration')
  })
  broker.on('connackSent', function () {
    t.fail('should no connack be sent')
  })
  broker.on('clientError', function (client, err) {
    t.error(err)
  })

  connect(setup(broker))

  evt.on('AuthenticateBegin', function (client) {
    t.equal(broker.connectedClients, 0)
    client.close()
  })
})

test('client is closed before authorizePublish returns', function (t) {
  t.plan(3)

  const evt = new EventEmitter()
  const broker = aedes({
    authorizePublish: (client, packet, done) => {
      evt.emit('AuthorizePublishBegin', client)
      // simulate latency writing to persistent store.
      setTimeout(function () {
        done()
        evt.emit('AuthorizePublishEnd', client)
      }, 2000)
    }
  })

  broker.on('clientError', function (client, err) {
    t.equal(err.message, 'connection closed')
  })

  const s = connect(setup(broker))
  s.inStream.write({
    cmd: 'publish',
    topic: 'hello',
    payload: 'world',
    qos: 1,
    messageId: 10,
    retain: false
  })

  evt.on('AuthorizePublishBegin', function (client) {
    t.equal(broker.connectedClients, 1)
    client.close()
  })
  evt.on('AuthorizePublishEnd', function (client) {
    t.equal(broker.connectedClients, 0)
    broker.close()
  })
})

test('close client when its socket is closed', function (t) {
  t.plan(4)

  const broker = aedes()
  t.teardown(broker.close.bind(broker))

  const subscriber = connect(setup(broker))

  subscribe(t, subscriber, 'hello', 1, function () {
    subscriber.inStream.end()
    subscriber.conn.on('close', function () {
      t.equal(broker.connectedClients, 0, 'no connected client')
    })
  })
})

test('multiple clients subscribe same topic, and all clients still receive message except the closed one', function (t) {
  t.plan(5)

  const mqtt = require('mqtt')
  const broker = aedes()

  let client2

  t.teardown(() => {
    client2.end()
    broker.close()
    server.close()
  })

  const server = require('net').createServer(broker.handle)
  const port = 1883
  server.listen(port)
  broker.on('clientError', function (client, err) {
    t.error(err)
  })

  const _sameTopic = 'hello'

  // client 1
  const client1 = mqtt.connect('mqtt://localhost', { clientId: 'client1', resubscribe: false, reconnectPeriod: -1 })
  client1.on('message', () => {
    t.fail('client1 receives message')
  })

  client1.subscribe(_sameTopic, { qos: 0, retain: false }, () => {
    t.pass('client1 sub callback')
    // stimulate closed socket by users
    client1.stream.destroy()

    // client 2
    client2 = mqtt.connect('mqtt://localhost', { clientId: 'client2', resubscribe: false })
    client2.on('message', () => {
      t.pass('client2 receives message')
      t.equal(broker.connectedClients, 1)
    })
    client2.subscribe(_sameTopic, { qos: 0, retain: false }, () => {
      t.pass('client2 sub callback')

      // pubClient
      const pubClient = mqtt.connect('mqtt://localhost', { clientId: 'pubClient' })
      pubClient.publish(_sameTopic, 'world', { qos: 0, retain: false }, () => {
        t.pass('pubClient publish event')
        pubClient.end()
      })
    })
  })
})


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

const write = require('../write')

function ClientPacketStatus (client, packet) {
  this.client = client
  this.packet = packet
}

function PubComp (packet) {
  this.cmd = 'pubcomp'
  this.messageId = packet.messageId
}

const pubrelActions = [
  pubrelGet,
  pubrelPublish,
  pubrelWrite,
  pubrelDel
]
function handlePubrel (client, packet, done) {
  client.broker._series(
    new ClientPacketStatus(client, packet),
    pubrelActions, {}, done)
}

function pubrelGet (arg, done) {
  const persistence = this.client.broker.persistence
  persistence.incomingGetPacket(this.client, this.packet, reply)

  function reply (err, packet) {
    arg.packet = packet
    done(err)
  }
}

function pubrelPublish (arg, done) {
  this.client.broker.publish(arg.packet, this.client, done)
}

function pubrelWrite (arg, done) {
  write(this.client, new PubComp(arg.packet), done)
}

function pubrelDel (arg, done) {
  const persistence = this.client.broker.persistence
  persistence.incomingDelPacket(this.client, arg.packet, done)
}

module.exports = handlePubrel


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

function handlePuback (client, packet, done) {
  const persistence = client.broker.persistence
  persistence.outgoingClearMessageId(client, packet, function (err, origPacket) {
    client.broker.emit('ack', origPacket, client)
    done(err)
  })
}

module.exports = handlePuback


<|start_filename|>benchmarks/throughputCounterQoS1.js<|end_filename|>
#! /usr/bin/env node

const mqtt = require('mqtt')

const client = mqtt.connect({ port: 1883, host: 'localhost', clean: true, encoding: 'binary', keepalive: 0 })
const interval = 5000

let counter = 0

function count () {
  console.log('received/s', counter / interval * 1000)
  counter = 0
}

setInterval(count, interval)

client.on('connect', function () {
  this.subscribe('test', { qos: 1 })
})

client.handleMessage = function (packet, done) {
  counter++
  done()
}

client.on('offline', function () {
  console.log('offline')
})

client.on('error', function () {
  console.log('reconnect!')
  client.stream.end()
})


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

const write = require('../write')
const pingResp = {
  cmd: 'pingresp'
}

function handlePing (client, packet, done) {
  client.broker.emit('ping', packet, client)
  write(client, pingResp, done)
}

module.exports = handlePing


<|start_filename|>test/keep-alive.js<|end_filename|>
'use strict'

const { test } = require('tap')
const eos = require('end-of-stream')
const Faketimers = require('@sinonjs/fake-timers')
const { setup, connect, noError } = require('./helper')
const aedes = require('../')

test('supports pingreq/pingresp', function (t) {
  t.plan(1)

  const s = noError(connect(setup()))
  t.teardown(s.broker.close.bind(s.broker))

  s.broker.on('keepaliveTimeout', function (client) {
    t.fail('keep alive should not timeout')
  })

  s.inStream.write({
    cmd: 'pingreq'
  })

  s.outStream.on('data', function (packet) {
    t.equal(packet.cmd, 'pingresp', 'the response is a pingresp')
  })
})

test('supports keep alive disconnections', function (t) {
  t.plan(2)

  const clock = Faketimers.install()
  const s = connect(setup(), { keepalive: 1 })
  t.teardown(s.broker.close.bind(s.broker))

  s.broker.on('keepaliveTimeout', function (client) {
    t.pass('keep alive timeout')
  })
  eos(s.conn, function () {
    t.pass('waits 1 and a half the keepalive timeout')
  })

  setTimeout(() => {
    clock.uninstall()
  }, 1.5)
  clock.tick(1.5)
})

test('supports keep alive disconnections after a pingreq', function (t) {
  t.plan(3)

  const clock = Faketimers.install()
  const s = connect(setup(), { keepalive: 1 })
  t.teardown(s.broker.close.bind(s.broker))

  eos(s.conn, function () {
    t.pass('waits 1 and a half the keepalive timeout')
  })
  s.broker.on('keepaliveTimeout', function (client) {
    t.pass('keep alive timeout')
  })
  s.outStream.on('data', function (packet) {
    t.equal(packet.cmd, 'pingresp', 'the response is a pingresp')
  })
  setTimeout(() => {
    s.inStream.write({
      cmd: 'pingreq'
    })
    clock.uninstall()
  }, 1)
  clock.tick(3)
})

test('disconnect if a connect does not arrive in time', function (t) {
  t.plan(2)

  const clock = Faketimers.install()
  const s = setup(aedes({
    connectTimeout: 500
  }))
  t.teardown(s.broker.close.bind(s.broker))

  s.client.on('error', function (err) {
    t.equal(err.message, 'connect did not arrive in time')
  })
  eos(s.conn, function () {
    t.pass('waits waitConnectTimeout before ending')
  })
  setTimeout(() => {
    clock.uninstall()
  }, 1000)
  clock.tick(1000)
})


<|start_filename|>test/helper.js<|end_filename|>
'use strict'

const duplexify = require('duplexify')
const mqtt = require('mqtt-connection')
const { through } = require('../lib/utils')
const util = require('util')
const aedes = require('../')

const parseStream = mqtt.parseStream
const generateStream = mqtt.generateStream
let clients = 0

function setup (broker) {
  const inStream = generateStream()
  const outStream = parseStream()
  const conn = duplexify(outStream, inStream)

  broker = broker || aedes()

  return {
    client: broker.handle(conn),
    conn: conn,
    inStream: inStream,
    outStream: outStream,
    broker: broker
  }
}

function connect (s, opts, connected) {
  s = Object.create(s)
  s.outStream = s.outStream.pipe(through(filter))

  opts = opts || {}

  opts.cmd = 'connect'
  opts.protocolId = opts.protocolId || 'MQTT'
  opts.protocolVersion = opts.protocolVersion || 4
  opts.clean = !!opts.clean
  opts.clientId = opts.clientId || 'my-client-' + clients++
  opts.keepalive = opts.keepalive || 0

  s.inStream.write(opts)

  return s

  function filter (packet, enc, cb) {
    if (packet.cmd !== 'publish') {
      delete packet.topic
      delete packet.payload
    }

    // using setImmediate to wait for connected to be fired
    // setup also needs to return first
    if (packet.cmd !== 'connack') {
      setImmediate(this.push.bind(this, packet))
    } else if (connected && packet.returnCode === 0) {
      setImmediate(connected, packet)
    }
    cb()
  }
}

function noError (s, t) {
  s.broker.on('clientError', function (client, err) {
    if (err) throw err
    t.notOk(err, 'must not error')
  })

  return s
}

function subscribe (t, subscriber, topic, qos, done) {
  subscriber.inStream.write({
    cmd: 'subscribe',
    messageId: 24,
    subscriptions: [{
      topic: topic,
      qos: qos
    }]
  })

  subscriber.outStream.once('data', function (packet) {
    t.equal(packet.cmd, 'suback')
    t.same(packet.granted, [qos])
    t.equal(packet.messageId, 24)

    if (done) {
      done(null, packet)
    }
  })
}

// subs: [{topic:, qos:}]
function subscribeMultiple (t, subscriber, subs, expectedGranted, done) {
  subscriber.inStream.write({
    cmd: 'subscribe',
    messageId: 24,
    subscriptions: subs
  })

  subscriber.outStream.once('data', function (packet) {
    t.equal(packet.cmd, 'suback')
    t.same(packet.granted, expectedGranted)
    t.equal(packet.messageId, 24)

    if (done) {
      done(null, packet)
    }
  })
}

module.exports = {
  setup: setup,
  connect: connect,
  noError: noError,
  subscribe: subscribe,
  subscribeMultiple: subscribeMultiple,
  delay: util.promisify(setTimeout)
}


<|start_filename|>benchmarks/pingpong.js<|end_filename|>
#! /usr/bin/env node

const mqtt = require('mqtt')
const convertHrtime = require('convert-hrtime')
const mode = require('compute-mode')
const client = mqtt.connect({ port: 1883, host: 'localhost', clean: true, keepalive: 0 })
const interval = 5000

let sent = 0
const latencies = []

function count () {
  console.log('sent/s', sent / interval * 1000)
  sent = 0
}

setInterval(count, interval)

function publish () {
  sent++
  client.publish('test', JSON.stringify(process.hrtime()), { qos: 1 })
}

function subscribe () {
  client.subscribe('test', { qos: 1 }, publish)
}

client.on('connect', subscribe)
client.on('message', publish)
client.on('message', function (topic, payload) {
  const sentAt = JSON.parse(payload)
  const diff = process.hrtime(sentAt)
  latencies.push(convertHrtime(diff).ms)
})

client.on('offline', function () {
  console.log('offline')
})

client.on('error', function () {
  console.log('reconnect!')
  client.stream.end()
})

process.on('SIGINT', function () {
  const total = latencies.reduce(function (acc, num) {
    return acc + num
  })
  console.log('total', total)
  console.log('average', total / latencies.length)
  console.log('mode', mode(latencies))
  process.exit(0)
})
