wip

pynais/junction.py

@@ -495,7 +495,7 @@ async def open(self):
 
         asyncio.ensure_future(self.dispatch())
 
-        await self.mqtt.subscribe([('hb/' + self.topic, QOS_1)])
+        await self.mqtt.subscribe([(self.topic, QOS_1)])
 
         asyncio.ensure_future(self.mqtt_reader_handler())
 
@@ -517,6 +517,7 @@ async def dispatch(self):
 
             Route messages enqueued by related endpoints
         """
+        topic = 'hb/' + self.topic
         while True:
             LOG.debug("MQTT: dequeueing from %s", self)
             message = await self.msg_queue.get()
@@ -525,8 +526,8 @@ async def dispatch(self):
             #     message = bytes(message, 'utf-8')
 
             LOG.debug("topic: |%s| - message to deliver: |%r|",
-                      self.topic, message)
-            await self.mqtt.publish(self.topic, message.data)
+                      topic, message)
+            await self.mqtt.publish(topic, message.data)
 
 
 class TcpLine(Line):

recipes/cc3200/mqtt_leds.py

@@ -29,16 +29,19 @@ async def main(color, value, topic):
     try:
         while True:
             message = await mqtt.deliver_message()
-
-            obj = ns.unmarshall(message.publish_packet.payload.data)
-            if ns.is_ack(obj):
-                break
+            payload = message.publish_packet.payload.data
+            if ns.is_protobuf(payload):
+                obj = ns.unmarshall(payload)
+                if ns.is_ack(obj):
+                    break
+            else:
+                LOG.debug(">> %s", payload)
 
         await mqtt.unsubscribe([sub_topic])
         await mqtt.disconnect()
-        
-    except ClientException as ce:
-        LOG.error("Client exception: %s" % ce)
+
+    except ClientException as cli_exc:
+        LOG.error("Client exception: %s", cli_exc)
 
 
 @click.command()