implement incoming fragmented packets for v0 and v1

prudpv0/src/server.rs

@@ -38,6 +38,7 @@ pub struct InternalConnection<C: CryptoInstance> {
     server_packet_counter: u16,
     client_packet_counter: u16,
     unacknowledged_packets: HashMap<u16, Arc<Vec<u8>>>,
+    packet_buffer: Vec<u8>,
     packet_queue: HashMap<u16, (Instant, PRUDPV0Packet<Vec<u8>>)>,
 }
 pub struct Connection<C: CryptoInstance> {
@@ -282,6 +283,7 @@ impl<C: Crypto> Server<C> {
                 server_packet_counter: 1,
                 unacknowledged_packets: HashMap::new(),
                 packet_queue: HashMap::new(),
+                packet_buffer: vec![],
             }),
         });
 
@@ -368,9 +370,16 @@ impl<C: Crypto> Server<C> {
             };
 
             conn.crypto_instance.decrypt_incoming(payload);
-
-            res.target.send(payload.to_owned()).await;
+            conn.packet_buffer.extend_from_slice(payload);
             conn.client_packet_counter += 1;
+            if packet.fragment_id().unwrap() != 0 {
+                info!("handeling fragmented packet");
+                continue;
+            }
+
+            res.target
+                .send(std::mem::take(&mut conn.packet_buffer))
+                .await;
         }
         info!("finished handeling packets, dropping inner connection");
         drop(conn);

prudpv1/src/prudp/socket.rs

@@ -50,6 +50,7 @@ struct InternalConnection<E: CryptoHandlerConnectionInstance> {
     socket: Arc<UdpSocket>,
     packet_queue: HashMap<u16, PRUDPV1Packet>,
     last_packet_time: Instant,
+    partial_packet: Vec<u8>,
     unacknowleged_packets: Vec<(Instant, PRUDPV1Packet)>,
 }
 
@@ -431,6 +432,7 @@ impl<T: CryptoHandler> InternalSocket<T> {
             packet_queue: Default::default(),
             last_packet_time: Instant::now(),
             unacknowleged_packets: Vec::new(),
+            partial_packet: Vec::new(),
             supported_function_version,
         };
 
@@ -573,11 +575,24 @@ impl<T: CryptoHandler> InternalSocket<T> {
         while let Some(mut packet) = conn.packet_queue.remove(&counter) {
             conn.crypto_handler_instance
                 .decrypt_incoming(packet.header.substream_id, &mut packet.payload[..]);
-
-            conn.data_sender.send(packet.payload).await.ok();
-
+            conn.partial_packet
+                .extend_from_slice(&mut packet.payload[..]);
             conn.reliable_client_counter = conn.reliable_client_counter.overflowing_add(1).0;
             counter = conn.reliable_client_counter;
+            if packet.options.iter().any(|v| {
+                if let FragmentId(f) = v {
+                    *f != 0
+                } else {
+                    false
+                }
+            }) {
+                println!("handeling fragmented packet");
+                continue;
+            }
+
+            let packet = std::mem::take(&mut conn.partial_packet);
+
+            conn.data_sender.send(packet).await.ok();
         }
     }