Aedes   

Barebone MQTT server that can run on any stream server.

• Install
• Example
• API
• TODO
• Acknowledgements
• License

Install

To install aedes, simply use npm:

npm install aedes --save

Example

var aedes = require('aedes')()
var server = require('net').createServer(aedes.handle)
var port = 1883

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

TLS

var fs = require('fs')
var aedes = require('aedes')()

var options = {
  key: fs.readFileSync('YOUR_TLS_KEY_FILE.pem'),
  cert: fs.readFileSync('YOUR_TLS_CERT_FILE.pem')
}

var server = require('tls').createServer(options, aedes.handle)

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

API

• aedes()
• instance.handle()
• instance.subscribe()
• instance.publish()
• instance.unsubscribe()
• instance.authenticate()
• instance.authorizePublish()
• instance.authorizeSubscribe()
• instance.authorizeForward()
• instance.published()
• instance.close()
• Client
• client.id
• client.clean
• client.publish()
• client.subscribe()
• client.close()

---

aedes([opts])

Creates a new instance of Aedes.

Options:

• mq: an instance of MQEmitter, such as MQEmitterRedis or MQEmitterMongoDB
• persistence: an instance of AedesPersistence, such as aedes-persistence-redis, aedes-persistence-nedb or aedes-persistence-mongodb
• concurrency: the max number of messages delivered concurrently, defaults to 100
• heartbeatInterval: the interval at which the broker heartbeat is emitted, it used by other broker in the cluster, defaults to 60000 milliseconds
• connectTimeout: the max number of milliseconds to wait for the CONNECT packet to arrive, defaults to 30000 milliseconds
• authenticate: function used to authenticate clients, see instance.authenticate()
• authorizePublish: function used to authorize PUBLISH packets, see instance.authorizePublish()
• authorizeSubscribe: function used to authorize SUBSCRIBE packets, see instance.authorizeSubscribe()
• authorizeForward: function used to authorize forwarded packets, see instance.authorizeForward()
• published: function called when a new packet is published, see instance.published()

Events:

• client: when a new Client connects, arguments:
  1. client
• clientDisconnect: when a Client disconnects, arguments:
  1. client
• clientError: when a Client errors, arguments:
  1. client
  2. err
• keepaliveTimeout: when a Client keepalive times out, arguments:
  1. client
• publish: when a new packet is published, arguments:
  1. packet
  2. client, it will be null if the message is published using publish.
• ack: when a packet published to a client is delivered successfully with QoS 1 or QoS 2, arguments:
  1. packet
  2. client
• ping: when a Client sends a ping, arguments:
  1. packet
  2. client
• subscribe: when a client sends a SUBSCRIBE, arguments:
  1. subscriptions, as defined in the subscriptions property of the SUBSCRIBE packet.
  2. client
• unsubscribe: when a client sends a UNSUBSCRIBE, arguments:
  1. unsubscriptions, as defined in the subscriptions property of the UNSUBSCRIBE packet.
  2. client
• connackSent: when a CONNACK packet is sent to a client Client (happens after 'client'), arguments:
  1. client
• closed: when the broker is closed

---

instance.handle(duplex)

Handle the given duplex as a MQTT connection.

var aedes = require('./aedes')()
var server = require('net').createServer(aedes.handle)

---

instance.subscribe(topic, func(packet, cb), done)

After done is called, every time publish is invoked on the instance (and on any other connected instances) with a matching topic the func function will be called.

func needs to call cb after receiving the message.

It supports backpressure.

---

instance.publish(packet, done)

Publish the given packet to subscribed clients and functions. It supports backpressure.

A packet contains the following properties:

{
  cmd: 'publish',
  qos: 2,
  topic: 'test',
  payload: new Buffer('test'),
  retain: false
}

Only the topic property is mandatory. Both topic and payload can be Buffer objects instead of strings.

---

instance.unsubscribe(topic, func(packet, cb), done)

The reverse of subscribe.

---

instance.authenticate(client, username, password, done(err, successful))

It will be called when a new client connects. Override to supply custom authentication logic.

instance.authenticate = function (client, username, password, callback) {
  callback(null, username === 'matteo')
}

Other return codes can passed as follows :-

instance.authenticate = function (client, username, password, callback) {
  var error = new Error('Auth error')
  error.returnCode = 1
  callback(error, null)
}

The return code values and their responses which can be passed are given below:

• 1 - Unacceptable protocol version
• 2 - Identifier rejected
• 3 - Server unavailable
• 4 - Bad user name or password

---

instance.authorizePublish(client, packet, done(err))

It will be called when a client publishes a message. Override to supply custom authorization logic.

instance.authorizePublish = function (client, packet, callback) {
  if (packet.topic === 'aaaa') {
    return callback(new Error('wrong topic'))
  }

  if (packet.topic === 'bbb') {
    packet.payload = new Buffer('overwrite packet payload')
  }

  callback(null)
}

---

instance.authorizeSubscribe(client, pattern, done(err, pattern))

It will be called when a client subscribes to a topic. Override to supply custom authorization logic.

instance.authorizeSubscribe = function (client, sub, callback) {
  if (sub.topic === 'aaaa') {
    return callback(new Error('wrong topic'))
  }

  if (sub.topic === 'bbb') {
    // overwrites subscription
    sub.qos = sub.qos + 2
  }

  callback(null, sub)
}

To negate a subscription, set the subscription to null:

instance.authorizeSubscribe = function (client, sub, callback) {
  if (sub.topic === 'aaaa') {
    sub = null
  }

  callback(null, sub)
}

---

instance.authorizeForward(clientId, packet)

It will be called when a client is set to recieve a message. Override to supply custom authorization logic.

instance.authorizeForward = function (clientId, packet) {
  if (packet.topic === 'aaaa' && clientId === "I should not see this") {
    return null
    // also works with return undefined
  }

  if (packet.topic === 'bbb') {
    packet.payload = new Buffer('overwrite packet payload')
  }

  return packet
}

---

instance.published(packet, client, done())

It will be called after a message is published. client will be null for internal messages. Override to supply custom authorization logic.

---

instance.close([cb])

Disconnects all clients.

Events:

• closed, in case the broker is closed

---

Client

Classes for all connected clients.

Events:

• error, in case something bad happended

---

client#id

The id of the client, as specified by the CONNECT packet.

---

client#client

true if the client connected (CONNECT) with clean: true, false otherwise. Check the MQTT spec for what this means.

---

client#publish(message, [callback])

Publish the given message to this client. QoS 1 and 2 are fully respected, while the retained flag is not.

message is a PUBLISH packet.

callback  will be called when the message has been sent, but not acked.

---

client#subscribe(subscriptions, [callback])

Subscribe the client to the list of topics.

subscription can be:

1. a single object in the format { topic: topic, qos: qos }
2. an array of the above
3. a full subscribe packet, specifying a messageId will send suback to the client.

callback  will be called when the subscription is completed.

---

client#unsubscribe(unsubscriptions, [callback])

Subscribe the client to the list of topics.

subscription can be:

1. a single object in the format topic
2. an array of the above
3. a full unsubscribe packet, specifying a messageId will send suback to the client.

callback  will be called when the unsubscription is completed.

---

client#close([cb])

Disconnects the client

---

client presence

You can subscribe on the following $SYS topics to get client presence:

• $SYS/+/new/clients - will inform about new clients connections
• $SYS/+/disconnect/clients - will inform about client disconnections. The payload will contain the clientId of the connected/disconnected client

Todo

• QoS 0 support
• Retain messages support
• QoS 1 support
• QoS 2 support
• clean=false support
• Keep alive support
• Will messages must survive crash
• Authentication
• Events
• Wait a CONNECT packet only for X seconds
• Support a CONNECT packet without a clientId
• Disconnect other clients with the same client.id
• Write docs
• Support counting the number of offline clients and subscriptions
• Performance optimizations for QoS 1 and Qos 2
• Add client#publish() and client#subscribe()
• move the persistence in a separate module
• mongo persistence (external module)
• redis persistence (external module)
• leveldb persistence (external module)
• cluster support (external module)

Acknowledgements

This library is born after a lot of discussion with all Mosca users and how that was deployed in production. This addresses your concerns about performance and stability.

License

MIT