Do not forward or store QoS 1/2 on $SYS/# for # subs.

aedes.js

@@ -141,6 +141,7 @@ function enqueueOffline (_, done) {
 
   enqueuer.complete = done
   enqueuer.status = this
+  enqueuer.topic = packet.topic
 
   this.broker.persistence.subscriptionsByTopic(
     packet.topic,
@@ -152,6 +153,7 @@ function DoEnqueues () {
   this.next = null
   this.status = null
   this.complete = null
+  this.topic = null
 
   var that = this
 
@@ -166,8 +168,13 @@ function DoEnqueues () {
     } else {
       var complete = that.complete
 
+      if (that.topic.indexOf('$SYS') === 0) {
+        subs = subs.filter(removeSharp)
+      }
+
       that.status = null
       that.complete = null
+      that.topic = null
 
       broker._parallel(
         status,
@@ -178,6 +185,10 @@ function DoEnqueues () {
   }
 }
 
+function removeSharp (sub) {
+  return sub.topic !== '#'
+}
+
 function doEnqueue (sub, done) {
   this.broker.persistence.outgoingEnqueue(sub, this.packet, done)
 }

lib/handlers/subscribe.js

@@ -36,6 +36,16 @@ function authorize (sub, done) {
   client.broker.authorizeSubscribe(client, sub, done)
 }
 
+function blockSys (func) {
+  return function deliverSharp (packet, cb) {
+    if (packet.topic.indexOf('$SYS') === 0) {
+      cb()
+    } else {
+      func(packet, cb)
+    }
+  }
+}
+
 function subTopic (sub, done) {
   if (!sub) {
     this.granted.push(128)
@@ -56,6 +66,10 @@ function subTopic (sub, done) {
       break
   }
 
+  if (sub.topic === '#') {
+    func = blockSys(func)
+  }
+
   client.subscriptions[sub.topic] = sub.qos
   this.granted.push(sub.qos)
 

test/events.js

@@ -1,7 +1,11 @@
 'use strict'
 
 var test = require('tape').test
+var helper = require('./helper')
 var aedes = require('../')
+var setup = helper.setup
+var connect = helper.connect
+var subscribe = helper.subscribe
 
 test('publishes an hearbeat', function (t) {
   t.plan(3)
@@ -16,3 +20,49 @@ test('publishes an hearbeat', function (t) {
     broker.close(t.pass.bind(t, 'broker closes'))
   })
 })
+
+test('does not forward $SYS topics to # subscription', function (t) {
+  t.plan(4)
+  var s = connect(setup())
+
+  subscribe(t, s, '#', 0, function () {
+    s.outStream.once('data', function (packet) {
+      t.fail('no packet should be received')
+    })
+
+    s.broker.mq.emit({
+      cmd: 'publish',
+      topic: '$SYS/hello',
+      payload: 'world'
+    }, function () {
+      t.pass('nothing happened')
+    })
+  })
+})
+
+test('does not store $SYS topics to QoS 1 # subscription', function (t) {
+  t.plan(3)
+
+  var broker = aedes()
+  var opts = { clean: false, clientId: 'abcde' }
+  var s = connect(setup(broker), opts)
+
+  subscribe(t, s, '#', 1, function () {
+    s.inStream.end()
+
+    s.broker.publish({
+      cmd: 'publish',
+      topic: '$SYS/hello',
+      payload: 'world',
+      qos: 1
+    }, function () {
+      s = connect(setup(broker), { clean: false, clientId: 'abcde' }, function () {
+        t.end()
+      })
+
+      s.outStream.once('data', function (packet) {
+        t.fail('no packet should be received')
+      })
+    })
+  })
+})

test/qos1.js

@@ -194,7 +194,8 @@ test('remove stored subscriptions if connected with clean=true', function (t) {
 test('resend publish on non-clean reconnect QoS 1', function (t) {
   var broker = aedes()
   var publisher
-  var subscriber = connect(setup(broker), { clean: false, clientId: 'abcde' })
+  var opts = { clean: false, clientId: 'abcde' }
+  var subscriber = connect(setup(broker), opts)
   var expected = {
     cmd: 'publish',
     topic: 'hello',
@@ -221,7 +222,7 @@ test('resend publish on non-clean reconnect QoS 1', function (t) {
     publisher.outStream.once('data', function (packet) {
       t.equal(packet.cmd, 'puback')
 
-      subscriber = connect(setup(broker), { clean: false, clientId: 'abcde' })
+      subscriber = connect(setup(broker), opts)
 
       subscriber.outStream.once('data', function (packet) {
         subscriber.inStream.write({