Skip to content
Barebone MQTT broker that can run on any stream server, the node way
JavaScript TypeScript
Branch: master
Clone or download
gnought Optimize handle function (#297)
* Optimize handle function

* Be more realistic, use handle fn in a test

* Optimize handle fn again

* add assert in test, destory should not be called

* Replaced assert with t.fail
Latest commit 992cce1 Aug 14, 2019
Permalink
Type Name Latest commit message Commit time
Failed to load latest commit information.
benchmarks Added pingpong benchmark. Dec 11, 2015
lib Optimize handle function (#297) Aug 13, 2019
test Optimize handle function (#297) Aug 13, 2019
types Replaced tslint with @typescript-eslint, and format standard output (#… Jul 17, 2019
.coveralls.yml Removed service-name from .coveralls.yml. Dec 16, 2015
.gitignore Replace istanbul with nyc (#253) Jul 11, 2019
.travis.yml
LICENSE Update LICENSE (#278) Jul 22, 2019
README.md Update README.md Jul 29, 2019
aedes.js Expose current Aedes version, used in aedes-stats (#294) Aug 13, 2019
example.js Updated readme and example file Jan 24, 2018
package.json Added a unit test, check offline queue in QoS 1 (#264) Aug 3, 2019

README.md

Aedes

Build Status Dependencies Status devDependencies Status
Known Vulnerabilities Coverage Status NPM version NPM downloads

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
  • 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
    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

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.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.


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:

  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

Collaborators

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

You can’t perform that action at this time.