mod: server refactoring read-process-send

Author: eigen-value

src/server.h

@@ -13,10 +13,9 @@
 #include "SerialTransport.h"
 
 #define MAX_CALLBACKS   100
+#define RPC_BUFFER_SIZE  1024
 
 class RPCServer {
-    RpcDecoder<>* decoder = nullptr;
-    RpcFunctionDispatcher<MAX_CALLBACKS> dispatcher;
 
 public:
     RPCServer(ITransport& t) : decoder(&RpcDecoderManager<>::getDecoder(t)) {}
@@ -33,10 +32,83 @@ class RPCServer {
 
     void run() {
         decoder->decode();
-        decoder->process_requests(dispatcher);
+        read_rpc();
+        process_request();
+        send_response();
         //delay(1);
     }
 
+protected:
+    void read_rpc() {
+        _rpc_size = decoder->read_rpc(_rpc_buffer, RPC_BUFFER_SIZE);
+    }
+
+    void process_request() {
+        if (_rpc_size == 0) return;
+
+        MsgPack::Unpacker unpacker;
+
+        unpacker.clear();
+        if (!unpacker.feed(_rpc_buffer, _rpc_size)) {
+            _rpc_size = 0; // Reset size on error
+            return; // Error in unpacking
+        }
+
+        int msg_type;
+        int msg_id;
+        MsgPack::str_t method;
+        MsgPack::arr_size_t req_size;
+
+        if (!unpacker.deserialize(req_size, msg_type)) {
+            reset_rpc();
+            return; // Header not unpackable
+        }
+
+        if (msg_type == CALL_MSG && req_size.size() == REQUEST_SIZE) {
+            if (!unpacker.deserialize(msg_id, method)) {
+                reset_rpc();
+                return; // Method not unpackable
+            }
+        } else if (msg_type == NOTIFY_MSG && req_size.size() == NOTIFY_SIZE) {
+            if (!unpacker.deserialize(method)) {
+                reset_rpc();
+                return; // Method not unpackable
+            }
+        } else {
+            reset_rpc();
+            return; // Invalid request size/type
+        }
+
+        _rpc_type = msg_type;
+
+        MsgPack::arr_size_t resp_size(RESPONSE_SIZE);
+        res_packer.clear();
+        if (msg_type == CALL_MSG) res_packer.serialize(resp_size, RESP_MSG, msg_id);
+
+        dispatcher.call(method, unpacker, res_packer);
+        reset_rpc();
+
+    }
+
+    void send_response() {
+        if (res_packer.size() > 0) {
+            decoder->send(reinterpret_cast<const uint8_t*>(res_packer.data()), res_packer.size());
+        }
+    }
+
+private:
+    RpcDecoder<>* decoder = nullptr;
+    RpcFunctionDispatcher<MAX_CALLBACKS> dispatcher;
+    uint8_t _rpc_buffer[RPC_BUFFER_SIZE];
+    size_t _rpc_size = 0;
+    uint8_t _rpc_type = NO_MSG;
+    MsgPack::Packer res_packer;
+    
+    void reset_rpc() {
+        _rpc_size = 0;
+        _rpc_type = NO_MSG;
+    }
+
 };
 
 #endif //RPCLITE_SERVER_H