mod: recv_msg in chunks up to buffer capacity

Author: eigen-value

src/SerialTransport.h

@@ -8,32 +8,32 @@
 
 class SerialTransport: public ITransport {
 
-    Stream* uart;
+    Stream* _stream;
 
     public:
 
-        SerialTransport(Stream* _uart): uart(_uart){}
+        SerialTransport(Stream* stream): _stream(stream){}
 
         void begin(){}
 
         size_t write(const uint8_t* data, size_t size) override {
 
             for (size_t i=0; i<size; i++){
-                uart->write(data[i]);
+                _stream->write(data[i]);
             }
 
             return size;
         }
 
         size_t read(uint8_t* buffer, size_t size) override {
-            
+
             size_t r_size = 0;
 
-            while (uart->available()){
-                // if (r_size == size){
-                //     return r_size;  // ERROR
-                // }
-                buffer[r_size] = uart->read();
+            while (_stream->available()){
+                if (r_size == size){
+                    return r_size;
+                }
+                buffer[r_size] = _stream->read();
                 r_size++;
             }
 

src/client.h

@@ -32,32 +32,32 @@ class RPCClient {
 
         MsgPack::Unpacker unpacker;
 
-        delay(100);
+        // blocking call
+        while (!recv_msg(transport, unpacker)){delay(1)};
 
-        if (recv_msg(transport, unpacker)) {
-            
-            int r_msg_type;
-            int r_msg_id;
-            MsgPack::object::nil_t error;
+        int r_msg_type;
+        int r_msg_id;
+        MsgPack::object::nil_t error;
 
-            MsgPack::arr_size_t resp_size(4);
+        MsgPack::arr_size_t resp_size(4);
 
-            bool ok = unpacker.deserialize(resp_size, r_msg_type, r_msg_id, error, result);
+        bool ok = unpacker.deserialize(resp_size, r_msg_type, r_msg_id, error, result);
 
-            if (!ok){
-                //Serial.println("could not serialize resp");
-                return false;
-            }
+        if (!ok){
+            //Serial.println("could not serialize resp");
+            return false;
+        }
+
+        if (r_msg_id != msg_id){
+            //Serial.println("msg_id mismatch");
+            return false;
+        }
 
-            if (r_msg_id != msg_id){
-                //Serial.println("msg_id mismatch");
-                return false;
-            }
+        msg_id += 1;
 
-            msg_id += 1;
-            return true;
+        flush_buffer();
+        return true;
 
-        }
 
         return false;
     }

src/rpc.h

@@ -8,6 +8,12 @@
 #include "transport.h"
 #include "MsgPack.h"
 
+#define MAX_BUFFER_SIZE 1024
+#define CHUNK_SIZE      64
+
+uint8_t raw_buffer[MAX_BUFFER_SIZE] = {0};
+size_t raw_buffer_fill = 0;
+
 inline void send_msg(ITransport& transport, const MsgPack::bin_t<uint8_t>& buffer) {
     size_t size = buffer.size();
 
@@ -23,34 +29,30 @@ inline void send_msg(ITransport& transport, const MsgPack::bin_t<uint8_t>& buffe
 
 inline bool recv_msg(ITransport& transport, MsgPack::Unpacker& unpacker) {
 
-    size_t size = 256;
-    uint8_t raw_buffer[size] = {0};
+    uint8_t temp_buffer[CHUNK_SIZE] = {0};
 
-    int attempts = 0;
-
-    size_t bytes_read = 0;
-
-    while ((attempts<100) && (bytes_read == 0)) {
-        bytes_read = transport.read(raw_buffer, size);
-        attempts++;
-        delay(10);
-    }
+    size_t bytes_read = transport.read(temp_buffer, CHUNK_SIZE);
 
     if (bytes_read == 0){
         Serial.println("no bytes received");
         return false;
     }
 
-    unpacker.feed(raw_buffer, bytes_read);
-
-    Serial.println("got data");
-    for (size_t i=0; i<bytes_read; i++){
-        Serial.print(raw_buffer[i]);
-        Serial.print(" ");
+    if (raw_buffer_fill + bytes_read > MAX_BUFFER_SIZE){
+        // ERROR: trying to recover flushing the buffer
+        flush_buffer();
+        return false;
     }
-    Serial.println("");
 
-    return true;
+    memcpy(raw_buffer + raw_buffer_fill, temp_buffer, bytes_read);
+    raw_buffer_fill += bytes_read;
+
+    return unpacker.feed(raw_buffer, raw_buffer_fill);
+
+}
+
+inline void flush_buffer(){
+    raw_buffer_fill = 0;
 }
 
 #endif //RPCLITE_RPC_H