Aedes
Barebone MQTT server that can run on any stream server.
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:
mq
: an instance of MQEmitter,
such as MQEmitterRedis
or MQEmitterMongoDBpersistence
: an instance of AedesPersistence,
such as aedes-persistence-redis,
aedes-persistence-nedb
or aedes-persistence-mongodbconcurrency
: 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
millisecondsconnectTimeout
: the max number of milliseconds to wait for the CONNECT
packet to arrive, defaults to 30000
millisecondsauthenticate
: 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:
client
err
connectionError
When a Client connection errors and there is no clientId attached , arguments:
client
err
keepaliveTimeout
: when a Client keepalive times out, arguments:
client
publish
: when a new packet is published, arguments:
packet
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:
packet
client
ping
: when a Client sends a ping, arguments:
packet
client
subscribe
: when a client sends a SUBSCRIBE, arguments:
subscriptions
, as defined in the subscriptions
property of the
SUBSCRIBE
packet.client
unsubscribe
: when a client sends a UNSUBSCRIBE, arguments:
unsubscriptions
, as defined in the subscriptions
property of the
UNSUBSCRIBE
packet.client
connackSent
: when a CONNACK packet is sent to a client Client (happens after 'client'
), arguments:
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 version2
- Identifier rejected3
- Server unavailable4
- 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') {
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
}
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#clean
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:
- a single object in the format
{ topic: topic, qos: qos }
- an array of the above
- 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 :-
- a single object in the format
{ topic: topic, qos: qos }
- 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
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