Barebone MQTT server that can run on any stream server.
To install aedes, simply use npm:
npm install aedes --save
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)
})aedes()instance.handle()instance.subscribe()instance.publish()instance.unsubscribe()instance.authenticate()instance.authorizePublish()instance.authorizeSubscribe()instance.authorizeForward()instance.published()instance.close()Clientclient.idclient.cleanclient.publish()client.subscribe()client.close()
Creates a new instance of Aedes.
Options:
mq: an instance of MQEmitter, such as MQEmitterRedis or MQEmitterMongoDBpersistence: an instance of AedesPersistence, such as aedes-persistence-redis or aedes-persistence-mongodbconcurrency: the max number of messages delivered concurrently, defaults to100.heartbeatInterval: the interval at which the broker heartbeat is emitted, it used by other broker in the cluster, the default is60000milliseconds.connectTimeout: the max number of milliseconds to wait for the CONNECT packet to arrive, defaults to30000milliseconds.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:client
clientDisconnect: when a Client disconnects, arguments:client
clientError: when a Client errors, arguments:clienterr
keepaliveTimeout: when a Client keepalive times out, arguments:client
publish: when a new packet is published, arguments:packetclient, it will be null if the message is published usingpublish.
ack: when a packet published to a client is delivered successfully with QoS 1 or QoS 2, arguments:packetclient
ping: when a Client sends a ping, arguments:packetclient
subscribe: when a client sends a SUBSCRIBE, arguments:subscriptions, as defined in thesubscriptionsproperty of the SUBSCRIBE packet.client
unsubscribe: when a client sends a UNSUBSCRIBE, arguments:unsubscriptions, as defined in thesubscriptionsproperty of the UNSUBSCRIBE packet.client
connackSent: when a CONNACK packet is sent to a client Client (happens after'client'), arguments:client
Handle the given duplex as a MQTT connection.
var aedes = require('./aedes')()
var server = require('net').createServer(aedes.handle)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.
Publish the given packet to subscribed clients and functions. It supports backpressure.
A packet must contain the following properties:
{
cmd: 'publish',
qos: 2,
topic: 'test',
payload: new Buffer('test'),
retain: false
}Only the topic and properties are mandatory
Both topic and payload can be Buffer objects instead of strings.
The reverse of subscribe.
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')
}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 version2- Identifier rejected3- Server unavailable4- Bad user name or password
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)
}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 + 2
}
callback(null, sub)
}To negate a subscription, set the subscription to null:
instance.authorizeSubscribe = function (client, sub, cb) {
if (sub.topic === 'aaaa') {
sub = null
}
callback(null, sub)
}It will be called when a client is set to recieve a message. Override to supply custom authorization logic. In order to completely pre
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
}It will be after a message is published.
client will be null for internal messages.
Ovverride to supply custom authorization logic.
Disconnects all clients.
Classes for all connected clients.
Events:
error, in case something bad happended
The id of the client, as specified by the CONNECT packet.
true if the client connected (CONNECT) with clean: true, false
otherwise. Check the MQTT spec for what this means.
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.
Subscribe the client to the list of topics.
subscription can be:
- a single object in the format
{ topic: topic, qos: qos } - an array of the above
- a full subscribe
packet,
specifying a
messageIdwill send suback to the client.
callback will be called when the subscription is completed.
Disconnects the client
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
- 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()andclient#subscribe() - move the persistence in a separate module
- mongo persistence (external module)
- redis persistence (external module)
- leveldb persistence (external module)
- cluster support (external module)
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.
MIT