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)
})

API

• aedes()
• instance.handle()
• instance.subscribe()
• instance.publish()
• instance.unsubscribe()
• instance.authenticate()
• instance.authorizePublish()
• instance.authorizeSubscribe()
• 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.
• persistence: an instance of AedesPersistence.
• 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, the default is 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().
• 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
• 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
• 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

---

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. A packet must be valid for mqtt-packet.

It supports backpressure.

---

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. Ovverride to supply custom authentication logic.

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

---

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 publishes a message. Override to supply custom authorization logic.

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

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

  callback(null, sub)
}

---

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

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

---

instance.close([cb])

Disconnects all clients.

---

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#close([cb])

Disconnects the 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