feat(bridge): support incoming mosquitto bridge connections

README.md

@@ -81,7 +81,7 @@ Check Docker docs [here](https://github.com/moscajs/aedes-cli#docker)
 - [Dynamic Topics][dynamic_topics] Support
 - MQTT Bridge Support between aedes
 - [MQTT 5.0][mqttv5] _(not support yet)_
-- [Bridge Protocol][bridge_protocol] _(not support yet)_
+- [Bridge Protocol][bridge_protocol] _(incoming connections only)_
 
 ## Examples
 
@@ -95,6 +95,20 @@ Other info:
 
 - The repo [aedes-tests](https://github.com/moscajs/aedes-tests) is used to test aedes with clusters and different emitters/persistences. Check its source code to have a starting point on how to work with clusters
 
+## Bridge connections
+
+Normally, when publishing a message, the `retain` flag is consumed by Aedes and
+then set to `false`.  This is done for two reasons:
+
+- MQTT-3.3.1-9 states that it MUST set the RETAIN flag to 0 when a PUBLISH
+  Packet is sent to a Client because it matches an established subscription
+  regardless of how the flag was set in the message it received.
+- When operating as a cluster, only one Aedes node may store the packet
+
+Brokers that support the [Bridge Protocol][bridge_protocol] can connect to
+Aedes.  When connecting with this special protocol, subscriptions work as usual
+except that the `retain` flag in the packet is propagated as-is.
+
 ## Exensions
 
 - [aedes-logging]: Logging module for Aedes, based on Pino

aedes.js

@@ -170,7 +170,6 @@ function storeRetained (packet, done) {
 }
 
 function emitPacket (packet, done) {
-  packet.retain = false
   this.broker.mq.emit(packet, done)
 }
 
@@ -257,11 +256,46 @@ Aedes.prototype.publish = function (packet, client, done) {
   this._series(new PublishState(this, client, packet), publishFuncs, p, done)
 }
 
-Aedes.prototype.subscribe = function (topic, func, done) {
+function createRetainToFalseWrapper (broker, topic, func) {
+  func._retainToFalseWrappers = func._wrappers || {}
+  func._retainToFalseWrappers[topic] = func._retainToFalseWrappers[topic] || {}
+  func._retainToFalseWrappers[topic][broker] = function handlePacketSubscription (_packet, cb) {
+    _packet = new Packet(_packet, broker)
+    _packet.retain = false
+    func(_packet, cb)
+  }
+  func._retainToFalseWrappers[topic][broker]._freeWrapper = function freeWrapper () {
+    delete ((func._retainToFalseWrappers || {})[topic] || {})[broker]
+    if (Object.keys((func._retainToFalseWrappers || {})[topic] || {}).length === 0) {
+      delete func._retainToFalseWrappers[topic]
+    }
+    if (Object.keys(func._retainToFalseWrappers).length === 0) {
+      delete func._retainToFalseWrappers
+    }
+  }
+  return func._retainToFalseWrappers[topic][broker]
+}
+
+Aedes.prototype.subscribe = function (topic, func, rap, done) {
+  if (done === undefined && typeof rap === 'function') {
+    done = rap
+    rap = false
+  }
+  if (!rap) {
+    func = createRetainToFalseWrapper(this, topic, func)
+  }
   this.mq.on(topic, func, done)
 }
 
 Aedes.prototype.unsubscribe = function (topic, func, done) {
+  if (func) {
+    func = ((func._retainToFalseWrappers || {})[topic] || {})[this] || func
+  }
+
+  if (func && func._freeWrapper) {
+    func._freeWrapper()
+  }
+
   this.mq.removeListener(topic, func, done)
 }
 

docs/Aedes.md

@@ -20,7 +20,7 @@
   - [Event: connackSent](#event-connacksent)
   - [Event: closed](#event-closed)
   - [aedes.handle (stream)](#aedeshandle-stream)
-  - [aedes.subscribe (topic, deliverfunc, callback)](#aedessubscribe-topic-deliverfunc-callback)
+  - [aedes.subscribe (topic, deliverfunc\[, retainAsPublised\], callback)](#aedessubscribe-topic-deliverfunc-callback)
   - [aedes.unsubscribe (topic, deliverfunc, callback)](#aedesunsubscribe-topic-deliverfunc-callback)
   - [aedes.publish (packet, callback)](#aedespublish-packet-callback)
   - [aedes.close ([callback])](#aedesclose-callback)
@@ -176,16 +176,19 @@ const aedes = require('./aedes')()
 const server = require('net').createServer(aedes.handle)
 ```
 
-## aedes.subscribe (topic, deliverfunc, callback)
+## aedes.subscribe (topic, deliverfunc\[, retainAsPublised\], callback)
 
 - topic: `<string>`
+- retainAsPublished: `<boolean>`
 - deliverfunc: `<Function>` `(packet, cb) => void`
   - packet: `<aedes-packet>` & [`PUBLISH`][PUBLISH]
   - cb: `<Function>`
 - callback: `<Function>`
 
 Directly subscribe a `topic` in server side. Bypass [`authorizeSubscribe`](#handler-authorizesubscribe-client-subscription-callback)
 
+If `retainAsPublished` is `true` (default `false`), the retain flag in the packet is passed to `deliverFunc` as it was published.  Otherwise, it is set to `false` before being passed to `deliverFunc`.
+
 The `topic` and `deliverfunc` is a compound key to differentiate the uniqueness of its subscription pool. `topic` could be the one that is existed, in this case `deliverfunc` will be invoked as well as `SUBSCRIBE` does.
 
 `deliverfunc` supports backpressue.

lib/handlers/subscribe.js

@@ -23,9 +23,23 @@ function SubAck (packet, granted) {
   this.granted = granted
 }
 
-function Subscription (qos, func) {
+function Subscription (qos, func, rh, rap, nl) {
   this.qos = qos
   this.func = func
+
+  // retain-handling indicates how retained messages should be
+  // handled when a new subscription is created
+  // (see [MQTT-3.3.1-9] through [MQTT-3.3.1-11])
+  this.rh = rh
+
+  // retain-as-published indicates whether to leave the retain flag as-is (true)
+  // or to clear it before sending to subscriptions (false) default false
+  // (see [MQTT-3.3.1-12] through [MQTT-3.3.1-13])
+  this.rap = rap
+
+  // no-local indicates that a client should not receive its own
+  // messages (see [MQTT-3.8.3-3])
+  this.nl = nl
 }
 
 function SubscribeState (client, packet, restore, finish) {
@@ -36,10 +50,13 @@ function SubscribeState (client, packet, restore, finish) {
   this.subState = []
 }
 
-function SubState (client, packet, granted) {
+function SubState (client, packet, granted, rh, rap, nl) {
   this.client = client
   this.packet = packet
   this.granted = granted
+  this.rh = rh
+  this.rap = rap
+  this.nl = nl
 }
 
 // if same subscribed topic in subs array, we pick up the last one
@@ -66,7 +83,7 @@ function handleSubscribe (client, packet, restore, done) {
 }
 
 function doSubscribe (sub, done) {
-  const s = new SubState(this.client, this.packet, sub.qos)
+  const s = new SubState(this.client, this.packet, sub.qos, sub.rh, sub.rap, sub.nl)
   this.subState.push(s)
   this.actions.call(s, sub, done)
 }
@@ -119,6 +136,9 @@ function addSubs (sub, done) {
   const broker = client.broker
   const topic = sub.topic
   const qos = sub.qos
+  const rh = this.rh
+  const rap = this.rap
+  const nl = this.nl
   let func = qos > 0 ? client.deliverQoS : client.deliver0
 
   // [MQTT-4.7.2-1]
@@ -127,12 +147,12 @@ function addSubs (sub, done) {
   }
 
   if (!client.subscriptions[topic]) {
-    client.subscriptions[topic] = new Subscription(qos, func)
-    broker.subscribe(topic, func, done)
-  } else if (client.subscriptions[topic].qos !== qos) {
+    client.subscriptions[topic] = new Subscription(qos, func, rh, rap, nl)
+    broker.subscribe(topic, func, rap, done)
+  } else if (client.subscriptions[topic].qos !== qos || client.subscriptions[topic].rh !== rh || client.subscriptions[topic].rap !== rap || client.subscriptions[topic].nl !== nl) {
     broker.unsubscribe(topic, client.subscriptions[topic].func)
-    client.subscriptions[topic] = new Subscription(qos, func)
-    broker.subscribe(topic, func, done)
+    client.subscriptions[topic] = new Subscription(qos, func, rh, rap, nl)
+    broker.subscribe(topic, func, rap, done)
   } else {
     done()
   }

package.json

@@ -121,7 +121,7 @@
     "fastseries": "^2.0.0",
     "hyperid": "^2.0.5",
     "mqemitter": "^4.4.0",
-    "mqtt-packet": "^6.7.0",
+    "mqtt-packet": "^6.9.0",
     "readable-stream": "^3.6.0",
     "retimer": "^2.0.0",
     "reusify": "^1.0.4",

test/client-pub-sub.js

@@ -451,6 +451,89 @@ test('subscribe a client programmatically', function (t) {
   })
 })
 
+test('subscribe a client programmatically clears retain', function (t) {
+  t.plan(3)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  const expected = {
+    cmd: 'publish',
+    topic: 'hello',
+    payload: Buffer.from('world'),
+    dup: false,
+    length: 12,
+    qos: 0,
+    retain: false
+  }
+
+  broker.on('client', function (client) {
+    client.subscribe({
+      topic: 'hello',
+      qos: 0
+    }, function (err) {
+      t.error(err, 'no error')
+
+      broker.publish({
+        topic: 'hello',
+        payload: Buffer.from('world'),
+        qos: 0,
+        retain: true
+      }, function (err) {
+        t.error(err, 'no error')
+      })
+    })
+  })
+
+  const s = connect(setup(broker))
+
+  s.outStream.once('data', function (packet) {
+    t.deepEqual(packet, expected, 'packet matches')
+  })
+})
+
+test('subscribe a bridge programmatically keeps retain', function (t) {
+  t.plan(3)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  const expected = {
+    cmd: 'publish',
+    topic: 'hello',
+    payload: Buffer.from('world'),
+    dup: false,
+    length: 12,
+    qos: 0,
+    retain: true
+  }
+
+  broker.on('client', function (client) {
+    client.subscribe({
+      topic: 'hello',
+      qos: 0,
+      rap: true
+    }, function (err) {
+      t.error(err, 'no error')
+
+      broker.publish({
+        topic: 'hello',
+        payload: Buffer.from('world'),
+        qos: 0,
+        retain: true
+      }, function (err) {
+        t.error(err, 'no error')
+      })
+    })
+  })
+
+  const s = connect(setup(broker))
+
+  s.outStream.once('data', function (packet) {
+    t.deepEqual(packet, expected, 'packet matches')
+  })
+})
+
 test('subscribe throws error when QoS > 0', function (t) {
   t.plan(3)
 

test/retain.js

@@ -76,6 +76,38 @@ test('retain messages', function (t) {
   publisher.inStream.write(expected)
 })
 
+test('retain messages only once in cluster setup', function (t) {
+  t.plan(3)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  const expected = {
+    cmd: 'publish',
+    topic: 'hello',
+    payload: Buffer.from('world'),
+    qos: 0,
+    dup: false,
+    length: 12,
+    retain: true
+  }
+
+  const subscriberFunc = function (packet, cb) {
+    cb()
+    setImmediate(function () {
+      t.equal(packet.retain, false, 'packet must not have retain')
+      broker.unsubscribe('hello', subscriberFunc, function () {
+        t.notOk(subscriberFunc._retainToFalseWrappers, 'wrapper function must be freed')
+      })
+    })
+  }
+
+  broker.subscribe('hello', subscriberFunc, function () {
+    t.ok(((subscriberFunc._retainToFalseWrappers || {}).hello || {})[broker], 'wrapper function must be created')
+    broker.publish(expected)
+  })
+})
+
 test('avoid wrong deduping of retain messages', function (t) {
   t.plan(7)