Aedes

Barebone MQTT server that can run on any stream server.

QoS 0	QoS 1	QoS 2	auth	bridge	$SYS topics	SSL	dynamic topics	cluster	websockets	plugin system	MQTT 5
✔	✔	✔	✔	?	✔	✔	✔	✔	✔	✔	✘

• Install
• Example
• API
• TODO
• Plugins
• Collaborators
• Developers
• Acknowledgements
• Mosca Vs Aedes
• Contributors
• 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')
})

WEBSOCKETS

var aedes = require('./aedes')()
var server = require('net').createServer(aedes.handle)
var httpServer = require('http').createServer()
var ws = require('websocket-stream')
var port = 1883
var 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)
})

API

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

---

aedes([opts])

Creates a new instance of Aedes.

Options:

• mq: an instance of MQEmitter, check plugins for more mqemitters options. Used to share messages between multiple brokers instances (ex: clusters)
• persistence: an instance of AedesPersistence, check plugins for more persistence options. It's used to store QoS > 1, retained, will packets and subscriptions in memory or on disk (if not specified default persistence is in memory)
• concurrency: the max number of messages delivered concurrently, defaults to 100
• queueLimit: the max number of messages queued while client is waiting to connect, defaults to 42. If the number is exceeded connectionError is thrown with error Client queue limit reached
• 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
• id: id used to identify this broker instance in $SYS messages, defaults to uuidv4()
• decodeProtocol: function called when a valid buffer is received, see instance.decodeProtocol()
• preConnect: function called when a valid CONNECT is received, see instance.preConnect()
• 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 successfully connects and register itself to server, [connackSent event will be come after], arguments:
  1. client
• clientReady: when a new Client received all its offline messages, it is ready, arguments:
  1. client
• clientDisconnect: when a Client disconnects, arguments:
  1. client
• clientError: when a Client errors, arguments:
  1. client
  2. err
• connectionError When a Client connection errors and there is no clientId attached , 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. It is by design that the broker heartbeat will be on publish event, in this case client is null
• ack: when a packet published to a client is delivered successfully with QoS 1 or QoS 2, arguments:
  1. packet, this will be the original PUBLISH packet in QoS 1, and PUBREL in QoS 2
  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, arguments:
  1. packet
  2. client
• closed: when the broker is closed

---

new aedes.Server([opts])

Same as aedes(opts). Creates a new instance of Aedes. This variant is useful with TypeScript or ES modules.

---

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.decodeProtocol(client, buffer)

It will be called when aedes instance trustProxy is true and that it receives a first valid buffer from client. client object state is in default and its connected state is false. Use aedes-protocol-decoder to parse https headers (x-real-ip | x-forwarded-for) and proxy protocol v1 and v2 to retrieve information in client.connDetails. Override to supply custom protocolDecoder logic, if it returns an object with data property, this property will be parsed as an mqtt-packet.

instance.decodeProtocol = function(client, buffer) {
  var protocol = yourDecoder(client, buffer)
  return protocol
}

---

instance.preConnect(client, done(err, successful))

It will be called when aedes instance receives a first valid CONNECT packet from client. client object state is in default and its connected state is false. Any values in CONNECT packet (like clientId, clean flag, keepalive) will pass to client object after this call. Override to supply custom preConnect logic. Some use cases:

1. Rate Limit / Throttle by client.conn.remoteAddress
2. Check instance.connectedClient to limit maximum connections
3. IP blacklisting

instance.preConnect = function(client, callback) {
  callback(null, client.conn.remoteAddress === '::1') {
}

instance.preConnect = function(client, callback) {
  callback(new Error('connection error'), client.conn.remoteAddress !== '::1') {
}

---

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 (client, packet) {
  if (packet.topic === 'aaaa' && client.id === "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, defaults to 'aedes_' + shortid()

---

client#clean

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

---

client#conn

The client's connection stream object.

In the case of net.createServer brokers, it's the connection passed to the connectionlistener function by node's net.createServer API.

In the case of websocket-stream brokers, it's the stream argument passed to the handle function described in that library's documentation.

---

client#req

The HTTP Websocket upgrade request object passed to websocket broker's handle function by the websocket-stream library.

If your clients are connecting to aedes via websocket and you need access to headers or cookies, you can get them here. NOTE: this property is only present for websocket connections.

---

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(topicObjects, [callback])

Unsubscribe the client to the list of topics.

The topic objects can be as follows :-

1. a single object in the format { topic: topic, qos: qos }
2. an array of the above

callback  will be called when the unsubscriptions are 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

Plugins

• aedes-persistence: In-memory implementation of an Aedes persistence
  • aedes-persistence-mongodb: MongoDB persistence for Aedes
  • aedes-persistence-redis: Redis persistence for Aedes
  • aedes-persistence-level: LevelDB persistence for Aedes
  • aedes-persistence-nedb: NeDB persistence for Aedes
• mqemitter: An Opinionated Message Queue with an emitter-style API
  • mqemitter-redis: Redis-powered mqemitter
  • mqemitter-mongodb: Mongodb based mqemitter
  • mqemitter-child-process: Share the same mqemitter between a hierarchy of child processes
  • mqemitter-cs: Expose a MQEmitter via a simple client/server protocol
  • mqemitter-p2p: A P2P implementation of MQEmitter, based on HyperEmitter and a Merkle DAG
  • mqemitter-aerospike: Aerospike mqemitter based on @mcollina 's mqemitter
• aedes-logging: Logging module for Aedes, based on Pino
• aedes-stats: Stats for Aedes
• aedes-protocol-decoder: Protocol decoder for Aedes MQTT Broker

Collaborators

• Gavin D'mello
• Behrad Zari
• Gnought

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.

Developers

Want to contribute? Check our list of features/bugs here

Mosca vs Aedes

Example benchmark test with 1000 clients sending 5000 QoS 1 messsages. Used mqtt-benchmark with command:

mqtt-benchmark --broker tcp://localhost:1883 --clients 1000 --qos 1 --count 5000

Aedes

========= TOTAL (1000) =========
Total Ratio:                 1.000 (5000000/5000000)
Total Runtime (sec):         178.495
Average Runtime (sec):       177.845
Msg time min (ms):           0.077
Msg time max (ms):           199.805
Msg time mean mean (ms):     35.403
Msg time mean std (ms):      0.042
Average Bandwidth (msg/sec): 28.115
Total Bandwidth (msg/sec):   28114.678

Mosca

========= TOTAL (1000) =========
Total Ratio:                 1.000 (5000000/5000000)
Total Runtime (sec):         264.934
Average Runtime (sec):       264.190
Msg time min (ms):           0.070
Msg time max (ms):           168.116
Msg time mean mean (ms):     52.629
Msg time mean std (ms):      0.074
Average Bandwidth (msg/sec): 18.926
Total Bandwidth (msg/sec):   18925.942

Contributors

Backers

Thank you to all our backers! 🙏 [Become a backer]

Sponsors

Support this project by becoming a sponsor. Your logo will show up here with a link to your website. [Become a sponsor]

License

MIT