udp protocol: add serde support

.github/workflows/ci.yml

@@ -66,6 +66,8 @@ jobs:
       run: cargo test --verbose --profile "test-fast" --workspace
     - name: Run tests (aquatic_udp with io_uring)
       run: cargo test --verbose --profile "test-fast" -p aquatic_udp --features "io-uring"
+    - name: Run tests (aquatic_udp_protocol with serde)
+      run: cargo test --verbose --profile "test-fast" -p aquatic_udp_protocol --features "serde"
 
   test-file-transfers:
     runs-on: ubuntu-latest

crates/peer_id/Cargo.toml

@@ -20,5 +20,6 @@ compact_str = "0.7"
 hex = "0.4"
 regex = "1"
 serde = { version = "1", features = ["derive"] }
+serde_with = { version = "3", features = ["hex"] }
 quickcheck = { version = "1", optional = true }
 zerocopy = { version = "0.7", features = ["derive"] }

crates/peer_id/src/lib.rs

@@ -3,8 +3,10 @@ use std::{borrow::Cow, fmt::Display, sync::OnceLock};
 use compact_str::{format_compact, CompactString};
 use regex::bytes::Regex;
 use serde::{Deserialize, Serialize};
+use serde_with::{hex::Hex, serde_as};
 use zerocopy::{AsBytes, FromBytes, FromZeroes};
 
+#[serde_as]
 #[derive(
     Debug,
     Clone,
@@ -21,7 +23,8 @@ use zerocopy::{AsBytes, FromBytes, FromZeroes};
     FromZeroes,
 )]
 #[repr(transparent)]
-pub struct PeerId(pub [u8; 20]);
+#[serde(transparent)]
+pub struct PeerId(#[serde_as(as = "Hex")] pub [u8; 20]);
 
 impl PeerId {
     pub fn client(&self) -> PeerClient {

crates/udp/src/common.rs

@@ -250,17 +250,23 @@ mod tests {
 
         let f = PeerStatus::from_event_and_bytes_left;
 
-        assert_eq!(Stopped, f(AnnounceEvent::Stopped, NumberOfBytes::new(0)));
-        assert_eq!(Stopped, f(AnnounceEvent::Stopped, NumberOfBytes::new(1)));
-
-        assert_eq!(Seeding, f(AnnounceEvent::Started, NumberOfBytes::new(0)));
-        assert_eq!(Leeching, f(AnnounceEvent::Started, NumberOfBytes::new(1)));
-
-        assert_eq!(Seeding, f(AnnounceEvent::Completed, NumberOfBytes::new(0)));
-        assert_eq!(Leeching, f(AnnounceEvent::Completed, NumberOfBytes::new(1)));
-
-        assert_eq!(Seeding, f(AnnounceEvent::None, NumberOfBytes::new(0)));
-        assert_eq!(Leeching, f(AnnounceEvent::None, NumberOfBytes::new(1)));
+        assert_eq!(Stopped, f(AnnounceEvent::Stopped, NumberOfBytes(0.into())));
+        assert_eq!(Stopped, f(AnnounceEvent::Stopped, NumberOfBytes(1.into())));
+
+        assert_eq!(Seeding, f(AnnounceEvent::Started, NumberOfBytes(0.into())));
+        assert_eq!(Leeching, f(AnnounceEvent::Started, NumberOfBytes(1.into())));
+
+        assert_eq!(
+            Seeding,
+            f(AnnounceEvent::Completed, NumberOfBytes(0.into()))
+        );
+        assert_eq!(
+            Leeching,
+            f(AnnounceEvent::Completed, NumberOfBytes(1.into()))
+        );
+
+        assert_eq!(Seeding, f(AnnounceEvent::None, NumberOfBytes(0.into())));
+        assert_eq!(Leeching, f(AnnounceEvent::None, NumberOfBytes(1.into())));
     }
 
     // Assumes that announce response with maximum amount of ipv6 peers will
@@ -273,24 +279,24 @@ mod tests {
 
         let peers = ::std::iter::repeat(ResponsePeer {
             ip_address: Ipv6AddrBytes(Ipv6Addr::new(1, 1, 1, 1, 1, 1, 1, 1).octets()),
-            port: Port::new(1),
+            port: Port(1.into()),
         })
         .take(config.protocol.max_response_peers)
         .collect();
 
         let response = Response::AnnounceIpv6(AnnounceResponse {
             fixed: AnnounceResponseFixedData {
-                transaction_id: TransactionId::new(1),
-                announce_interval: AnnounceInterval::new(1),
-                seeders: NumberOfPeers::new(1),
-                leechers: NumberOfPeers::new(1),
+                transaction_id: TransactionId(1.into()),
+                announce_interval: AnnounceInterval(1.into()),
+                seeders: NumberOfPeers(1.into()),
+                leechers: NumberOfPeers(1.into()),
             },
             peers,
         });
 
         let mut buf = Vec::new();
 
-        response.write(&mut buf).unwrap();
+        response.write_bytes(&mut buf).unwrap();
 
         println!("Buffer len: {}", buf.len());
 

crates/udp/src/workers/socket/mio.rs

@@ -198,7 +198,7 @@ impl SocketWorker {
                     }
 
                     let src = CanonicalSocketAddr::new(src);
-                    let request_parsable = match Request::from_bytes(
+                    let request_parsable = match Request::parse_bytes(
                         &self.buffer[..bytes_read],
                         self.config.protocol.max_scrape_torrents,
                     ) {
@@ -431,7 +431,7 @@ impl SocketWorker {
     ) {
         let mut buffer = Cursor::new(&mut buffer[..]);
 
-        if let Err(err) = response.write(&mut buffer) {
+        if let Err(err) = response.write_bytes(&mut buffer) {
             ::log::error!("failed writing response to buffer: {:#}", err);
 
             return;

crates/udp/src/workers/socket/storage.rs

@@ -155,8 +155,8 @@ mod tests {
             }
 
             let request = ScrapeRequest {
-                transaction_id: TransactionId::new(t),
-                connection_id: ConnectionId::new(c),
+                transaction_id: TransactionId(t.into()),
+                connection_id: ConnectionId(c.into()),
                 info_hashes,
             };
 
@@ -191,9 +191,9 @@ mod tests {
                         (
                             i,
                             TorrentScrapeStatistics {
-                                seeders: NumberOfPeers::new((info_hash.0[0]) as i32),
-                                leechers: NumberOfPeers::new(0),
-                                completed: NumberOfDownloads::new(0),
+                                seeders: NumberOfPeers(((info_hash.0[0]) as i32).into()),
+                                leechers: NumberOfPeers(0.into()),
+                                completed: NumberOfDownloads(0.into()),
                             },
                         )
                     })

crates/udp/src/workers/socket/uring/recv_helper.rs

@@ -138,7 +138,7 @@ impl RecvHelper {
 
         let addr = CanonicalSocketAddr::new(addr);
 
-        let request = Request::from_bytes(msg.payload_data(), self.max_scrape_torrents)
+        let request = Request::parse_bytes(msg.payload_data(), self.max_scrape_torrents)
             .map_err(|err| Error::RequestParseError(err, addr))?;
 
         Ok((request, addr))

crates/udp/src/workers/socket/uring/send_buffers.rs

@@ -196,7 +196,7 @@ impl SendBuffer {
 
         let mut cursor = Cursor::new(&mut self.bytes[..]);
 
-        match response.write(&mut cursor) {
+        match response.write_bytes(&mut cursor) {
             Ok(()) => {
                 self.iovec.iov_len = cursor.position() as usize;