feat: recode with agnostic packet parsing

Author: eigen-value

src/client.h

@@ -11,6 +11,7 @@
 
 class RPCClient {
     RpcDecoder<>* decoder = nullptr;
+    int _waiting_msg_id;
 
 public:
     RpcError lastError;
@@ -31,23 +32,46 @@ class RPCClient {
     template<typename RType, typename... Args>
     bool call(const MsgPack::str_t method, RType& result, Args&&... args) {
 
-        int msg_id;
-        if (!decoder->send_call(CALL_MSG, method, msg_id, std::forward<Args>(args)...)){
+        if(!send_rpc(method, std::forward<Args>(args)...)) {
+            lastError.code = GENERIC_ERR;
+            lastError.traceback = "Failed to send RPC call";
+            return false;
         }
 
-        RpcError error;
         // blocking call
-        while (!decoder->get_response(msg_id, result, error)){
+        while (!get_response(result)){
             decoder->decode();
             //delay(1);
         }
 
-        lastError.code = error.code;
-        lastError.traceback = error.traceback;
+        return (lastError.code == NO_ERR);
+
+    }
+
+protected:
+    template<typename... Args>
+    bool send_rpc(const MsgPack::str_t method, Args&&... args) {
+        int msg_id;
+        if (decoder->send_call(CALL_MSG, method, msg_id, std::forward<Args>(args)...)) {
+            _waiting_msg_id = msg_id;
+            return true;
+        }
+        return false;
+    }
 
-        return (error.code == NO_ERR);
+    template<typename RType>
+    bool get_response(RType& result) {
+        RpcError tmp_error;
+        decoder->decode();
 
+        if (decoder->get_response(_waiting_msg_id, result, tmp_error)) {
+            lastError.code = tmp_error.code;
+            lastError.traceback = tmp_error.traceback;
+            return true;
+        }
+        return false;
     }
+
 };
 
 #endif //RPCLITE_CLIENT_H

src/decoder.h

@@ -8,14 +8,9 @@
 
 using namespace RpcUtils::detail;
 
-#define NO_MSG          -1
-#define CALL_MSG        0
-#define RESP_MSG        1
-#define NOTIFY_MSG      2
 
-#define REQUEST_SIZE    4
-#define RESPONSE_SIZE   4
-#define NOTIFY_SIZE     3
+
+#define MIN_RPC_BYTES   4
 
 #define MAX_BUFFER_SIZE 1024
 #define CHUNK_SIZE      32
@@ -56,118 +51,45 @@ class RpcDecoder {
     template<typename RType>
     bool get_response(const int msg_id, RType& result, RpcError& error) {
 
-        if (packet_type()!=RESP_MSG) return false;
+        if (!packet_incoming() || packet_type()!=RESP_MSG) return false;
 
         MsgPack::Unpacker unpacker;
+        unpacker.clear();
 
-        size_t bytes_checked = 0;
+        if (!unpacker.feed(_raw_buffer, get_packet_size())) return false;
 
-        while (bytes_checked < _bytes_stored) {
-            bytes_checked++;
-            unpacker.clear();
-            if (!unpacker.feed(_raw_buffer, bytes_checked)) continue;
-            MsgPack::arr_size_t resp_size;
-            int resp_type;
-            int resp_id;
-            if (!unpacker.deserialize(resp_size, resp_type, resp_id)) continue;
-            if (resp_size.size() != RESPONSE_SIZE) continue;
-            if (resp_type != RESP_MSG) continue;
-            if (resp_id != msg_id) continue;
-
-            MsgPack::object::nil_t nil;
-            if (unpacker.unpackable(nil)){  // No error
-                if (!unpacker.deserialize(nil, result)) continue;
-            } else {                        // RPC returned an error
-                if (!unpacker.deserialize(error, nil)) continue;
-            }
-            consume(bytes_checked);
-            return true;
-        }
-        return false;
-    }
-
-    template<typename RType>
-    bool send_response(const int msg_id, const RpcError& error, const RType& result) {
-        MsgPack::Packer packer;
-        MsgPack::arr_size_t resp_size(RESPONSE_SIZE);
-        MsgPack::object::nil_t nil;
+        MsgPack::arr_size_t resp_size;
+        int resp_type;
+        int resp_id;
 
-        packer.clear();
-        packer.serialize(resp_size, RESP_MSG, msg_id);
+        if (!unpacker.deserialize(resp_size, resp_type, resp_id)) return false;
+        if (resp_size.size() != RESPONSE_SIZE) return false;
+        if (resp_type != RESP_MSG) return false;
+        if (resp_id != msg_id) return false;
 
-        if (error.code == NO_ERR){
-            packer.serialize(nil, result);
-        } else {
-            packer.serialize(error, nil);
+        MsgPack::object::nil_t nil;
+        if (unpacker.unpackable(nil)){  // No error
+            if (!unpacker.deserialize(nil, result)) return false;
+        } else {                        // RPC returned an error
+            if (!unpacker.deserialize(error, nil)) return false;
         }
 
-        return send(reinterpret_cast<const uint8_t*>(packer.data()), packer.size()) == packer.size();
+        consume(get_packet_size());
+        return true;
 
     }
 
-    template<size_t N>
-    void process_requests(RpcFunctionDispatcher<N>& dispatcher) {
-        if (_packet_type!=CALL_MSG && _packet_type!=NOTIFY_MSG) return;
-
-        MsgPack::Unpacker unpacker;
-        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;
+    bool send_response(const MsgPack::Packer& packer) {
+        return send(reinterpret_cast<const uint8_t*>(packer.data()), packer.size()) == packer.size();
+    }
 
-            int msg_type;
-            int msg_id;
-            MsgPack::str_t method;
-            MsgPack::arr_size_t req_size;
-
-            if (!unpacker.deserialize(req_size, msg_type)) continue;
-            // todo HANDLE MALFORMED CLIENT REQ ERRORS
-            if ((req_size.size() == REQUEST_SIZE) && (msg_type == CALL_MSG)){
-                if (!unpacker.deserialize(msg_id, method)) continue;
-                if (unpacker.size() < REQUEST_SIZE + 1) continue;                       // there must be at least 5 indices
-            } else if ((req_size.size() == NOTIFY_SIZE) && (msg_type == NOTIFY_MSG)) {
-                if (!unpacker.deserialize(method)) continue;
-                if (unpacker.size() < NOTIFY_SIZE + 1) continue;                        // there must be at least 4 indices
-            } else if ((req_size.size() == RESPONSE_SIZE) && (msg_type == RESP_MSG)) {  // this should never happen but it's addressed to a client
-                break;
-            } else {
-                discard_packet();
-                break;
-            }
-            // Headers unpacked
-
-            MsgPack::arr_size_t resp_size(RESPONSE_SIZE);
-            packer.clear();
-            if (msg_type == CALL_MSG) 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());
-                    }   // 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());
-                }
-                consume(bytes_checked);
-                break;
-            }
+    size_t get_request(uint8_t* buffer, size_t buffer_size) {
 
+        if (packet_type() != CALL_MSG && packet_type() != NOTIFY_MSG) {
+            return 0; // No RPC
         }
 
+        return pop_packet(buffer, buffer_size);
     }
 
     void decode(){
@@ -187,52 +109,45 @@ class RpcDecoder {
 
     void parse_packet(){
 
-        if (packet_incoming() || _bytes_stored < 2){return;}
-
-        MsgPack::Unpacker unpacker;
-        unpacker.clear();
-        unpacker.feed(_raw_buffer, 2);
-
-        MsgPack::arr_size_t elem_size;
-        int type;
-        if (unpacker.deserialize(elem_size, type)){
-            _packet_type = type;
-        }
-
-    }
-
-    // Check if a packet is available
-    inline bool packet_incoming() const { return _packet_type >= CALL_MSG; }
-
-    int packet_type() const {return _packet_type;}
-
-    // Get the size of the next packet in the buffer (must be array contained, no other requirements)
-    size_t get_packet_size() {
+        if (packet_incoming()){return;}
 
         size_t bytes_checked = 0;
         size_t container_size;
+        int type;
         MsgPack::Unpacker unpacker;
 
         while (bytes_checked < _bytes_stored){
             bytes_checked++;
             unpacker.clear();
             if (!unpacker.feed(_raw_buffer, bytes_checked)) continue;
 
-            if (unpackArray(unpacker, container_size)) {
-                return bytes_checked;
+            if (unpackTypedArray(unpacker, container_size, type)) {
+
+                if (type != CALL_MSG && type != RESP_MSG && type != NOTIFY_MSG) {
+                    consume(bytes_checked);
+                    break; // Not a valid RPC type (could be type=WRONG_MSG)
+                }
+
+                if ((type == CALL_MSG && container_size != REQUEST_SIZE) || (type == RESP_MSG && container_size != RESPONSE_SIZE) || (type == NOTIFY_MSG && container_size != NOTIFY_SIZE)) {
+                    consume(bytes_checked);
+                    break; // Not a valid RPC format
+                }
+
+                _packet_type = type;
+                _packet_size = bytes_checked;
             } else {
                 continue;
             }
 
         }
 
-        return 0;
     }
 
-    // Discard the next (array) packet in the buffer, returns the number of bytes consumed.
-    size_t discard_packet() {
-        return consume(get_packet_size());
-    }
+    inline bool packet_incoming() const { return _packet_size >= MIN_RPC_BYTES; }
+
+    inline int packet_type() const { return _packet_type; }
+
+    size_t get_packet_size() const { return _packet_size;}
 
     inline size_t size() const {return _bytes_stored;}
 
@@ -241,6 +156,7 @@ class RpcDecoder {
     uint8_t _raw_buffer[BufferSize];
     size_t _bytes_stored = 0;
     int _packet_type = NO_MSG;
+    size_t _packet_size = 0;
     int _msg_id = 0;
 
     inline bool buffer_full() const { return _bytes_stored == BufferSize; }
@@ -251,7 +167,27 @@ class RpcDecoder {
         return _transport.write(data, size);
     }
 
-    // Consume the first 'size' bytes of the buffer, shifting remaining data forward
+    size_t pop_packet(uint8_t* buffer, size_t buffer_size) {
+
+        if (!packet_incoming()) return 0;
+
+        size_t packet_size = get_packet_size();
+        if (packet_size > buffer_size) return 0;
+
+        for (size_t i = 0; i < packet_size; i++) {
+            buffer[i] = _raw_buffer[i];
+        }
+
+        reset_packet();
+        return consume(packet_size);
+    }
+
+
+    void reset_packet() {
+        _packet_type = NO_MSG;
+        _packet_size = 0;
+    }
+
     size_t consume(size_t size) {
 
         if (size > _bytes_stored) return 0;
@@ -264,7 +200,6 @@ class RpcDecoder {
         }
 
         _bytes_stored = remaining_bytes;
-        _packet_type = NO_MSG;
 
         return size;
     }

src/dispatcher.h

@@ -40,7 +40,7 @@ class RpcFunctionDispatcher {
             }
         }
 
-        // handle not found
+        // handler not found
         MsgPack::object::nil_t nil;
         packer.serialize(RpcError(FUNCTION_NOT_FOUND_ERR, name), nil);
         return false;