Fix #30

hbmqtt/broker.py

@@ -556,9 +556,11 @@ class Broker:
             if '#' in a_filter and not a_filter.endswith('#'):
                 # [MQTT-4.7.1-2] Wildcard character '#' is only allowed as last character in filter
                 return 0x80
-            if '+' in a_filter and not wildcard_pattern.match(a_filter):
-                # [MQTT-4.7.1-3] + wildcard character must occupy entire level
-                return 0x80
+            if a_filter != "+":
+                if '+' in a_filter:
+                    if "/+" not in a_filter and "+/" not in a_filter:
+                        # [MQTT-4.7.1-3] + wildcard character must occupy entire level
+                        return 0x80
 
             qos = subscription[1]
             if 'max-qos' in self.config and qos > self.config['max-qos']:

tests/test_broker.py

@@ -388,6 +388,34 @@ class BrokerTest(unittest.TestCase):
         if future.exception():
             raise future.exception()
 
+    @patch('hbmqtt.broker.PluginManager')
+    def test_client_subscribe_invalid(self, MockPluginManager):
+        @asyncio.coroutine
+        def test_coro():
+            try:
+                broker = Broker(test_config, plugin_namespace="hbmqtt.test.plugins")
+                yield from broker.start()
+                self.assertTrue(broker.transitions.is_started())
+                sub_client = MQTTClient()
+                yield from sub_client.connect('mqtt://localhost')
+                ret = yield from sub_client.subscribe(
+                    [('+', QOS_0), ('+/tennis/#', QOS_0), ('sport+', QOS_0), ('sport/+/player1', QOS_0)])
+                self.assertEquals(ret, [QOS_0, QOS_0, 0x80, QOS_0])
+
+                yield from asyncio.sleep(0.1)
+                yield from sub_client.disconnect()
+                yield from asyncio.sleep(0.1)
+                yield from broker.shutdown()
+                self.assertTrue(broker.transitions.is_stopped())
+                future.set_result(True)
+            except Exception as ae:
+                future.set_exception(ae)
+
+        future = asyncio.Future(loop=self.loop)
+        self.loop.run_until_complete(test_coro())
+        if future.exception():
+            raise future.exception()
+
     @patch('hbmqtt.broker.PluginManager')
     def test_client_subscribe_publish_dollar_topic_1(self, MockPluginManager):
         @asyncio.coroutine