wip

Author: pavel-kirienko

libudpard/udpard.c

@@ -332,12 +332,12 @@ static byte_t* header_serialize(byte_t* const  buffer,
     return ptr;
 }
 
-static bool header_deserialize(const udpard_bytes_mut_t  dgram_payload,
-                               meta_t* const             out_meta,
-                               uint32_t* const           frame_index,
-                               uint32_t* const           frame_payload_offset,
-                               uint32_t* const           prefix_crc,
-                               udpard_bytes_mut_t* const out_payload)
+static bool header_deserialize(const udpard_bytes_mut_t dgram_payload,
+                               meta_t* const            out_meta,
+                               uint32_t* const          frame_index,
+                               uint32_t* const          frame_payload_offset,
+                               uint32_t* const          prefix_crc,
+                               udpard_bytes_t* const    out_payload)
 {
     UDPARD_ASSERT(out_payload != NULL);
     bool ok = (dgram_payload.size >= HEADER_SIZE_BYTES) && (dgram_payload.data != NULL) && //
@@ -1744,16 +1744,14 @@ bool udpard_rx_port_push(udpard_rx_t* const         rx,
     if (ok) {
         port->invoked = true;
         // Parse and validate the frame.
-        udpard_bytes_mut_t payload     = { 0 };
-        rx_frame_t         frame       = { 0 };
-        uint32_t           frame_index = 0;
-        uint32_t           offset_32   = 0;
-        const bool         frame_valid =
-          header_deserialize(datagram_payload, &frame.meta, &frame_index, &offset_32, &frame.base.crc, &payload);
+        rx_frame_t frame       = { 0 };
+        uint32_t   frame_index = 0;
+        uint32_t   offset_32   = 0;
+        const bool frame_valid = header_deserialize(
+          datagram_payload, &frame.meta, &frame_index, &offset_32, &frame.base.crc, &frame.base.payload);
         frame.base.offset = (size_t)offset_32;
-        (void)frame_index; // currently not used by this reassembler implementation.
-        UDPARD_ASSERT((frame.base.origin.data == datagram_payload.data) &&
-                      (frame.base.origin.size == datagram_payload.size));
+        (void)frame_index;                    // currently not used by this reassembler implementation.
+        frame.base.origin = datagram_payload; // Take ownership of the payload.
         // Process the frame.
         if (frame_valid) {
             if (frame.meta.topic_hash == port->topic_hash) {

tests/src/test_intrusive_header.c

@@ -28,7 +28,7 @@ static void test_header_v2(void)
     // Just verify deserialization works correctly
 
     meta_t             meta_out;
-    udpard_bytes_mut_t payload_out;
+    udpard_bytes_t     payload_out;
     uint32_t           frame_index          = 0;
     uint32_t           frame_payload_offset = 0;
     uint32_t           prefix_crc           = 0;
@@ -86,7 +86,7 @@ static void test_header_deserialize_edge_cases(void)
     };
 
     meta_t             meta_out;
-    udpard_bytes_mut_t payload_out;
+    udpard_bytes_t     payload_out;
     uint32_t           frame_index          = 0;
     uint32_t           frame_payload_offset = 0;
     uint32_t           prefix_crc           = 0;

tests/src/test_intrusive_rx.c

@@ -2876,6 +2876,45 @@ static void test_session_unordered(void)
     instrumented_allocator_reset(&alloc_payload);
 }
 
+// ---------------------------------------------  PORT  ---------------------------------------------
+
+/// Tests ports in ORDERED, UNORDERED, and STATELESS modes.
+/// The UNORDERED port is the p2p_port in the rx instance; the other modes are tested on separate port instances.
+/// All transfers are single-frame transfers for simplicity, since we already have dedicated lower-level tests.
+static void test_port(void)
+{
+    // Initialize the memory resources.
+    instrumented_allocator_t alloc_frag = { 0 };
+    instrumented_allocator_new(&alloc_frag);
+    const udpard_mem_resource_t mem_frag = instrumented_allocator_make_resource(&alloc_frag);
+
+    instrumented_allocator_t alloc_session = { 0 };
+    instrumented_allocator_new(&alloc_session);
+    const udpard_mem_resource_t mem_session = instrumented_allocator_make_resource(&alloc_session);
+
+    instrumented_allocator_t alloc_payload = { 0 };
+    instrumented_allocator_new(&alloc_payload);
+    const udpard_mem_resource_t mem_payload = instrumented_allocator_make_resource(&alloc_payload);
+    const udpard_mem_deleter_t  del_payload = instrumented_allocator_make_deleter(&alloc_payload);
+
+    const udpard_rx_memory_resources_t rx_mem = { .fragment = mem_frag, .session = mem_session };
+
+    // Initialize the shared RX instance.
+    const uint64_t local_uid = 0x6EC164169C3088B4ULL;
+    udpard_rx_t    rx;
+    TEST_ASSERT(udpard_rx_new(&rx, local_uid, rx_mem, &on_message, &on_collision, &on_ack_mandate));
+    callback_result_t cb_result = { 0 };
+    rx.user                     = &cb_result;
+    TEST_ASSERT_EQUAL(local_uid, rx.p2p_port.topic_hash);
+    TEST_ASSERT_EQUAL(UDPARD_REORDERING_WINDOW_UNORDERED, rx.p2p_port.reordering_window);
+
+    // TODO continue: init two ports, one ORDERED, one STATELESS. Feed frames from different remote nodes into each.
+    // Verify the callbacks: on message, on ack, on collision.
+    // Feed bad frames and verify they are rejected with no memory leaks.
+    (void)mem_payload;  // TODO: Will be used when test is completed
+    (void)del_payload;  // TODO: Will be used when test is completed
+}
+
 void setUp(void) {}
 
 void tearDown(void) {}
@@ -2897,5 +2936,7 @@ int main(void)
     RUN_TEST(test_session_ordered);
     RUN_TEST(test_session_unordered);
 
+    RUN_TEST(test_port);
+
     return UNITY_END();
 }