TBT feat: server using decoder. mod: wrapper handles incomplete parameters deserialization. todo discard_packet needed to recover malformed client requests

Author: eigen-value

examples/rpc_lite_server/rpc_lite_server.ino

@@ -1,7 +1,6 @@
 #include <Arduino_RPClite.h>
-#include <HardwareSerial.h>
 
-SerialTransport transport(&Serial2);
+SerialTransport transport(&Serial0);
 RPCServer server(transport);
 
 int add(int a, int b){
@@ -17,7 +16,7 @@ MsgPack::str_t loopback(MsgPack::str_t message){
 }
 
 void setup() {
-    Serial2.begin(115200);
+    Serial0.begin(115200);
     transport.begin();
     pinMode(LED_BUILTIN, OUTPUT);
     Serial.begin(9600);
@@ -37,5 +36,5 @@ void blink_before(){
 
 void loop() {
     blink_before();
-    server.loop();
+    server.run();
 }

src/decoder.h

@@ -1,7 +1,9 @@
 #ifndef RPCLITE_DECODER_H
 #define RPCLITE_DECODER_H
-#include "transport.h"
+
 #include "MsgPack.h"
+#include "transport.h"
+#include "dispatcher.h"
 
 
 #define NO_MSG          -1
@@ -113,6 +115,74 @@ class RpcDecoder {
 
     }
 
+    template<size_t N>
+    void process_requests(RpcFunctionDispatcher<N>& dispatcher) {
+        if (_packet_type!=CALL_MSG && _packet_type!=NOTIFY_MSG) return;
+
+        static MsgPack::Unpacker unpacker;
+        static MsgPack::Packer packer;
+
+        size_t bytes_checked = 0;
+
+        while (bytes_checked < _bytes_stored) {
+            bytes_checked++;
+            unpacker.clear();
+            if (!unpacker.feed(_raw_buffer, bytes_checked)) continue;
+
+            int msg_type;
+            int msg_id;
+            MsgPack::str_t method;
+            MsgPack::arr_size_t req_size;
+
+            if (!unpacker.deserialize(req_size, msg_type)) break;
+            // todo HANDLE MALFORMED CLIENT REQ ERRORS
+            if ((req_size.size() == REQUEST_SIZE) && (msg_type == CALL_MSG)){
+                if (!unpacker.deserialize(msg_id, method)) {
+                    discard_packet();
+                    break;
+                }
+            } else if ((req_size.size() == NOTIFY_SIZE) && (msg_type == NOTIFY_MSG)) {
+                if (!unpacker.deserialize(method)) {
+                    discard_packet();
+                    break;
+                }
+            } else if ((req_size.size() == RESPONSE_SIZE) && (msg_type == RESP_MSG)) {   // this should never happen
+                break;
+            } else {
+                discard_packet();
+                break;
+            }
+
+            MsgPack::arr_size_t resp_size(RESPONSE_SIZE);
+            packer.clear();
+            packer.serialize(resp_size, RESP_MSG, msg_id);
+            size_t headers_size = packer.size();
+
+            if (!dispatcher.call(method, unpacker, packer)) {
+                if (packer.size()==headers_size) {
+                    // Call didn't go through bc parameters are not ready yet
+                    continue;
+                } else {
+                    // something went wrong the call raised an error or the client issued a malformed request
+                    if (msg_type == CALL_MSG) {
+                        send(reinterpret_cast<const uint8_t*>(packer.data()), packer.size()) == packer.size();
+                    }   // if notification client will never know something went wrong
+                    discard_packet();   // agnostic pop
+                    break;
+                }
+            } else {
+                // all is well we can respond and pop the deserialized packet
+                if (msg_type == CALL_MSG){
+                    send(reinterpret_cast<const uint8_t*>(packer.data()), packer.size()) == packer.size();
+                }
+                pop_packet(bytes_checked);
+                break;
+            }
+
+        }
+
+    }
+
     void process(){
         if (advance()) parse_packet();
     }
@@ -192,6 +262,13 @@ class RpcDecoder {
         return size;
     }
 
+    size_t discard_packet() {
+        // todo recursive packet size measurement
+        // use pop_packet with size
+        // return popped bytes
+        return 0;
+    }
+
     inline size_t size() const {return _bytes_stored;}
 
 #ifdef DEBUG

src/decoder_manager.h

@@ -12,6 +12,7 @@
 template<size_t MaxTransports = RPCLITE_MAX_TRANSPORTS>
 class RpcDecoderManager {
 public:
+    // todo parametrize so the RpcDecoder returned has a user defined buffer size ?
     static RpcDecoder<>& getDecoder(ITransport& transport) {
         for (auto& entry : decoders_) {
             if (entry.transport == &transport) {

src/server.h

@@ -5,65 +5,33 @@
 #ifndef RPCLITE_SERVER_H
 #define RPCLITE_SERVER_H
 
-#include "rpc.h"
 #include "error.h"
 #include "wrapper.h"
 #include "dispatcher.h"
+#include "decoder.h"
+#include "decoder_manager.h"
 
 #define MAX_CALLBACKS   100
 
 class RPCServer {
     ITransport& transport;
+    RpcDecoder<>& decoder;
+    RpcFunctionDispatcher<MAX_CALLBACKS> dispatcher;
 
 public:
-    RPCServer(ITransport& t) : transport(t) {}
+    RPCServer(ITransport& t) : transport(t), decoder(RpcDecoderManager<>::getDecoder(t)) {}
 
     template<typename F>
     void bind(const MsgPack::str_t& name, F&& func){
         dispatcher.bind(name, func);
     }
 
-    void loop() {
-
-        if (!recv_msg(transport, unpacker)) return;
-
-        int msg_type;
-        int msg_id;
-        MsgPack::str_t method;
-        MsgPack::arr_size_t req_size;
-
-        if (!unpacker.deserialize(req_size, msg_type, msg_id, method)){
-            Serial.println("unable to deserialize a msg received");
-
-            for (size_t i=0; i<raw_buffer_fill; i++){
-                Serial.print(raw_buffer[i], HEX);
-                Serial.print(" ");
-            }
-            Serial.println(" ");
-
-            flush_buffer();
-            return;
-        } else {
-            Serial.print("calling method: ");
-            Serial.println(method);
-        }
-
-        MsgPack::arr_size_t resp_size(4);
-        packer.clear();
-        packer.serialize(resp_size, RESP_MSG, msg_id);
-
-        dispatcher.call(method, unpacker, packer);
-        flush_buffer();
-        send_msg(transport, packer.packet());
-
+    void run() {
+        decoder.process();
+        decoder.process_requests(dispatcher);
+        delay(1);
     }
 
-private:
-
-    RpcFunctionDispatcher<MAX_CALLBACKS> dispatcher;
-    MsgPack::Unpacker unpacker;
-    MsgPack::Packer packer;
-
 };
 
 #endif //RPCLITE_SERVER_H

src/wrapper.h

@@ -94,7 +94,10 @@ class RpcFunctionWrapper<R(Args...)>: public IFunctionWrapper {
             return false;
         }
 
-        auto args = deserialize_all<Args...>(unpacker);
+        //unpacker not ready if deserialization fails at this point
+        std::tuple<Args...> args;
+        if (!deserialize_all<Args...>(unpacker, args)) return false;
+
         if constexpr (std::is_void<R>::value){
             invoke_with_tuple(_func, args, arx::stdx::make_index_sequence<sizeof...(Args)>{});
             packer.serialize(nil, nil);
@@ -119,16 +122,31 @@ class RpcFunctionWrapper<R(Args...)>: public IFunctionWrapper {
 private:
     std::function<R(Args...)> _func;
 
+    template<size_t I = 0, typename... Ts>
+    inline typename std::enable_if<I == sizeof...(Ts), bool>::type
+    deserialize_tuple(MsgPack::Unpacker& unpacker, std::tuple<Ts...>& out) {
+        return true; // Base case
+    }
+
+    template<size_t I = 0, typename... Ts>
+    inline typename std::enable_if<I < sizeof...(Ts), bool>::type
+    deserialize_tuple(MsgPack::Unpacker& unpacker, std::tuple<Ts...>& out) {
+        if (!deserialize_single(unpacker, std::get<I>(out))) {
+            return false;
+        }
+        return deserialize_tuple<I+1>(unpacker, out); // Recursive unpacking
+    }
+
     template<typename... Ts>
-    std::tuple<Ts...> deserialize_all(MsgPack::Unpacker& unpacker) {
-        return std::make_tuple(deserialize_single<Ts>(unpacker)...);
+    bool deserialize_all(MsgPack::Unpacker& unpacker, std::tuple<Ts...>& values) {
+        return deserialize_tuple(unpacker, values);
     }
 
     template<typename T>
-    static T deserialize_single(MsgPack::Unpacker& unpacker) {
-        T value;
+    static bool deserialize_single(MsgPack::Unpacker& unpacker, T& value) {
+        if (!unpacker.unpackable(value)) return false;
         unpacker.deserialize(value);
-        return value;
+        return true;
     }
 };