Merge pull request 256dpi#14 from 256dpi/size-prefix

Size Prefix

Author: 256dpi

src/YunMQTTClient.cpp

@@ -51,13 +51,13 @@ boolean YunMQTTClient::connect(const char * clientId, const char * username, con
   this->process.readStringUntil('\n');
 
   // set will if available
-  if(this->willTopic != NULL) {
+  if(strlen(this->willTopic) > 0) {
     this->process.print("w:");
     this->process.print(this->willTopic);
-    if(this->willPayload != NULL) {
-      this->process.print(':');
-      this->process.print(this->willPayload);
-    }
+    this->process.print(':');
+    this->process.print(strlen(this->willPayload));
+    this->process.print(';');
+    this->process.print(this->willPayload);
     this->process.print('\n');
   }
 
@@ -74,11 +74,11 @@ boolean YunMQTTClient::connect(const char * clientId, const char * username, con
     this->process.print(':');
     this->process.print(password);
   }
-  this->process.print('\n');
+  this->process.print(";\n");
 
   // wait for answer
   String ret = this->process.readStringUntil('\n');
-  this->alive = ret.equals("a");
+  this->alive = ret.equals("a;");
 
   if(!this->alive) {
     this->process.close();
@@ -105,6 +105,8 @@ void YunMQTTClient::publish(const char * topic, const char * payload) {
   this->process.print("p:");
   this->process.print(topic);
   this->process.print(':');
+  this->process.print(strlen(payload));
+  this->process.print(';');
   this->process.print(payload);
   this->process.print('\n');
 }
@@ -117,7 +119,7 @@ void YunMQTTClient::subscribe(const char * topic) {
   // send subscribe request
   this->process.print("s:");
   this->process.print(topic);
-  this->process.print('\n');
+  this->process.print(";\n");
 }
 
 void YunMQTTClient::unsubscribe(String topic) {
@@ -128,26 +130,34 @@ void YunMQTTClient::unsubscribe(const char * topic) {
   // send unsubscribe request
   this->process.print("u:");
   this->process.print(topic);
-  this->process.print('\n');
+  this->process.print(";\n");
 }
 
 void YunMQTTClient::loop() {
   int av = this->process.available();
   if(av > 0) {
-    String ret = process.readStringUntil('\n');
+    String ret = process.readStringUntil(';');
 
     if(ret.startsWith("m")) {
       int startTopic = 2;
       int endTopic = ret.indexOf(':', startTopic + 1);
-      int startPayload = endTopic + 1;
-      int endPayload = ret.indexOf(':', startPayload + 1);
       String topic = ret.substring(startTopic, endTopic);
-      String payload = ret.substring(startPayload, endPayload);
-      messageReceived(topic, payload, (char*)payload.c_str(), payload.length());
+
+      int startPayloadLength = endTopic + 1;
+      int endPayloadLength = ret.indexOf(':', startPayloadLength + 1);
+      int payloadLength = ret.substring(startPayloadLength, endPayloadLength).toInt();
+
+      char buf[payloadLength+1];
+      process.readBytes(buf, payloadLength);
+      buf[payloadLength] = '\0';
+
+      messageReceived(topic, String(buf), buf, payloadLength);
     } else if(ret.startsWith("e")) {
       this->alive = false;
       this->process.close();
     }
+
+    process.readStringUntil('\n');
   }
 }
 
@@ -157,7 +167,7 @@ boolean YunMQTTClient::connected() {
 
 void YunMQTTClient::disconnect() {
   // send disconnect request
-  this->process.print("d\n");
+  this->process.print("d;\n");
 }
 
 #endif //ARDUINO_AVR_YUN

src/YunMQTTClient.h

@@ -13,8 +13,8 @@ class YunMQTTClient {
   Process process;
   const char * hostname;
   int port;
-  const char * willTopic = NULL;
-  const char * willPayload = NULL;
+  const char * willTopic = "";
+  const char * willPayload = "";
   boolean alive = false;
   boolean updateBridge();
 public:

yun/abi.md

@@ -5,16 +5,17 @@
 The following commands are exchanged between the python script and the arduino library:
 
    | Command     | Format
----|-------------|-----------------------------------------
--> | will        | `w:topic:(payload)`
--> | connect     | `c:host:port:id:(user):(pass)`
-<- | approved    | `a`
-<- | rejected    | `r`
--> | subscribe   | `s:(topic)`
--> | unsubscribe | `u:(topic)`
--> | publish     | `p:(topic):(data)`
-<- | message     | `m:(topic):(data)`
--> | disconnect  | `d`
-<- | closed      | `e`
+---|-------------|------------------------------------
+<- | boot        | `b;`
+-> | will        | `w:topic:payload_len;(payload)`
+-> | connect     | `c:host:port:id:(user):(pass);`
+<- | approved    | `a;`
+<- | rejected    | `r;`
+-> | subscribe   | `s:topic;`
+-> | unsubscribe | `u:topic;`
+-> | publish     | `p:topic:payload_len;(payload)`
+<- | message     | `m:topic:payload_len;(payload)`
+-> | disconnect  | `d;`
+<- | closed      | `e;`
 
 All commands end with a standard line break `\n`.

yun/bridge.py

@@ -2,6 +2,10 @@
 import mqtt
 
 class Bridge:
+    COMMAND_END = b"\n"
+    ARGUMENTS_SEPERATOR = b":"
+    PAYLOAD_SEPERATOR = b";"
+
     # Constructor
     def __init__(self):
         self.client = None
@@ -10,19 +14,16 @@ def __init__(self):
 
     # Bridge Callbacks
     def on_connect(self, _, __, ___, rc):
-        if rc == 0:
-            self.send_command("a")
-        else:
-            self.send_command("r")
+        self.send_command("a;" if rc == 0 else "r;")
     def on_message(self, _, __, msg):
-        self.send_command("m:" + msg.topic + ":" + str(msg.payload))
+        self.send_command("m:" + msg.topic + ":" + str(len(msg.payload)) + ";" + str(msg.payload))
     def on_disconnect(self):
         self.client.loop_stop()
-        self.send_command("e")
+        self.send_command("e;")
 
     # Command Helpers
     def parse_command(self, line):
-        segments = line.split(":")
+        segments = line.split(self.ARGUMENTS_SEPERATOR)
         cmd = segments[0]
         remaining = segments[1:]
         if cmd == 'w':
@@ -37,51 +38,54 @@ def parse_command(self, line):
             self.do_publish(remaining)
         elif cmd == 'd':
             self.do_disconnect()
+        self.read_until(self.COMMAND_END)
     def send_command(self, line):
-        sys.stdout.write(line + "\n")
+        sys.stdout.write(line + self.COMMAND_END)
 
     # Command Handlers
     def do_will(self, args):
         self.will_topic = args[0]
-        if len(args) >= 2:
-            self.will_payload = args[1]
+        self.will_payload = self.read_chunk(int(args[1]))
 
     def do_connect(self, args):
-        if len(args) >= 3:
-            self.client = mqtt.Client(args[2])
-            self.client.on_connect = self.on_connect
-            self.client.on_message = self.on_message
-            if len(args) >= 5:
-                self.client.username_pw_set(args[3], args[4])
-            if len(self.will_topic) > 0:
-                if len(self.will_payload) > 0:
-                    self.client.will_set(self.will_topic, self.will_payload)
-                else:
-                    self.client.will_set(self.will_topic)
-            try:
-                self.client.connect(args[0], int(args[1]))
-                self.client.loop_start()
-            except:
-                self.send_command("r")
+        self.client = mqtt.Client(args[2])
+        self.client.on_connect = self.on_connect
+        self.client.on_message = self.on_message
+        if len(args) >= 5:
+            self.client.username_pw_set(args[3], args[4])
+        if len(self.will_topic) > 0:
+            self.client.will_set(self.will_topic, self.will_payload)
+        try:
+            self.client.connect(args[0], int(args[1]))
+            self.client.loop_start()
+        except:
+            self.send_command("r;")
 
     def do_subscribe(self, args):
-        if self.client and len(args) >= 1:
-            self.client.subscribe(args[0])
+        self.client.subscribe(args[0])
     def do_unsubscribe(self, args):
-        if self.client and len(args) >= 1:
-            self.client.unsubscribe(args[0])
+        self.client.unsubscribe(args[0])
     def do_publish(self, args):
-        if self.client and len(args) >= 2:
-            self.client.publish(args[0], args[1])
+        self.client.publish(args[0], self.read_chunk(int(args[1])))
     def do_disconnect(self):
-        if self.client:
-            self.client.disconnect()
+        self.client.disconnect()
 
     # Main
     def run(self):
-        self.send_command("ok")
+        self.send_command("b;")
         while True:
-            self.parse_command(sys.stdin.readline()[0:-1])
+            self.parse_command(self.read_until(self.PAYLOAD_SEPERATOR))
+
+    # Low Level Helpers
+    def read_until(self, end):
+        s = b""
+        c = sys.stdin.read(1)
+        while c not in end:
+            s += c
+            c = sys.stdin.read(1)
+        return s
+    def read_chunk(self, length):
+        return sys.stdin.read(length)
 
 # Main Loop
 Bridge().run()