/aedes

Nodejs mqtt broker.

Primary LanguageJavaScriptMIT LicenseMIT

Aedes  Build Status Coverage Status

Barebone MQTT server that can run on any stream server.

js-standard-style

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([opts])

Creates a new instance of Aedes.

Options:

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