aedes-persistence

aedes-persistence

The spec for an Aedes persistence, with abstract tests and a fast in-memory implementation.

• Install
• API
• Implement another persistence
• License

Install

To install aedes-persistence, simply use npm:

npm install aedes-persistence --save

API

• persistence()
• instance.storeRetained()
• instance.createRetainedStream()
• instance.createRetainedStreamCombi()
• instance.addSubscriptions()
• instance.removeSubscriptions()
• instance.subscriptionsByClient()
• instance.countOffline()
• instance.subscriptionsByTopic()
• instance.cleanSubscriptions()
• instance.outgoingEnqueue()
• instance.outgoingEnqueueCombi()
• instance.outgoingUpdate()
• instance.outgoingClearMessageId()
• instance.outgoingStream()
• instance.incomingStorePacket()
• instance.incomingGetPacket()
• instance.incomingDelPacket()
• instance.putWill()
• instance.getWill()
• instance.delWill()
• instance.streamWill()
• instance.getClientList()
• instance.destroy()

---

persistence([opts])

Creates a new instance of a persistence, that is already ready to operate. The default implementation is in-memory only.

---

instance.storeRetained(packet, callback(err))

Store a retained message, calls the callback when it was saved.

---

instance.createRetainedStream(pattern)

Return a stream that will load all retained messages matching the given pattern (according to the MQTT spec) asynchronously. Deprecated.

---

instance.createRetainedStreamCombi(patterns)

Return a stream that will load all retained messages matching given patterns (according to the MQTT spec) asynchronously.

---

instance.addSubscriptions(client, subscriptions, callback(err, client))

Add the given offline subscriptions for the given Client. The client must have connected with clean: false, as this is not checked here. This is called when a client issue a SUBSCRIBE packet.

subscriptions is in the same format of the subscribe property in the SUBSCRIBE packet:

[{
  topic: 'hello/world',
  qos: 1,
}, {
  topic: 'hello/#',
  qos: 2,
}]

---

instance.removeSubscriptions(client, subscriptions, callback(err, client))

The inverse of addSubscriptions but subscriptions is an array of topic names.

---

instance.subscriptionsByClient(client, callback(err, subscriptions, client))

Returns all the offline subscriptions for the given client. Called when a client with clean: false connects to restore its subscriptions.

subscriptions is in the same format of the subscribe property in the SUBSCRIBE packet:

[{
  topic: 'hello/world',
  qos: 1,
}, {
  topic: 'hello/#',
  qos: 2,
}]

---

instance.countOffline(cb(err, numOfSubscriptions, numOfClients))

Returns the number of offline subscriptions and the number of offline clients.

---

instance.subscriptionsByTopic(pattern, callback(err, subscriptions))

Returns all the offline subscriptions matching the given pattern. Called when a PUBLISH with qos: 1 or qos: 2 is received.

The subscriptions are in the format:

{
  clientId: client.id,
  topic: sub.topic,
  qos: sub.qos
}

---

instance.cleanSubscriptions(client, callback(err, client))

Removes all offline subscriptions for a given client.

---

instance.outgoingEnqueue(subscription, packet, callback(err))

Enqueue a potentially offline delivery. subscription is one of the objects returned by subscriptionsByTopic. Deprecated.

---

instance.outgoingEnqueueCombi(subscriptions, packet, callback(err))

Enqueue a potentially offline delivery. subscriptions is the whole subscriptions objects returned by subscriptionsByTopic.

---

instance.outgoingUpdate(client, packet, callback(err))

Called before a (potentially) offline packet is delivered, the caller should update the packet.messageId before updating.

---

instance.outgoingClearMessageId(client, packet, callback(err, packet))

Removes a packet with the given messageId (passing a PUBACK is ok) from the persistence. Passes back original packet to the callback.

---

instance.outgoingStream(client)

Return a stream that will load all offline messages for the given client asynchronously.

---

instance.incomingStorePacket(client, packet, cb(err, packet))

Store an incoming packet for the given client. Used for QoS 2.

---

instance.incomingGetPacket(client, packet, cb(err, packet))

Retrieve an incoming packet with the same messageId for the given client. Used for QoS 2.

---

instance.incomingDelPacket(client, packet, cb(err, packet))

Deletes incoming packet with the same messageId for the given client. Used for QoS 2.

---

instance.putWill(client, packet, cb(err))

Stores the will of a client. Used to support multi-broker environments and to not lose wills in case of a crash.

---

instance.getWill(client, packet, cb(err))

Retrieves the will of a client. Used to support multi-broker environments and to not lose wills in case of a crash.

---

instance.delWill(client, packet, cb(err))

Removes the will of a client. Used to support multi-broker environments and to not lose wills in case of a crash.

---

instance.streamWill(brokers)

Streams all the wills for the given brokers. The brokers are in the format:

{
  mybroker: {
    brokerId: 'mybroker'
  }
}

---

instance.getClientList(topic)

Returns a stream which has all the clientIds subscribed to the specified topic

instance.destroy(cb(err))

Destroy current persistence. Use callback cb(err) to catch errors if any

Implement another persistence

A persistence needs to pass all tests defined in ./abstract.js. You can import and use that test suite in the following manner:

const test = require('tape').test
const myperst = require('./')
const abs = require('aedes-persistence/abstract')

abs({
  test: test,
  persistence: myperst
})

If you require some async stuff before returning, a callback is also supported:

const test = require('tape').test
const myperst = require('./')
const abs = require('aedes-persistence/abstract')
const clean = require('./clean') // invented module

abs({
  test: test,
  buildEmitter: require('mymqemitter'), // optional
  persistence: function build (cb) {
    clean(function (err) {
      cb(err, myperst())
    })
  }
})

Collaborators

• Gnought

License

MIT