Subscription I/O

Cargo.toml

@@ -219,6 +219,7 @@ async-trait = "0.1.88"
 base64 = "0.22.1"
 bytes = { version = "1.10.1", features = ["serde"] }
 cached = { version = "0.54", default-features = false }
+compact_str = "0.7"
 corrosion = { path = "crates/corrosion" }
 eyre = "0.6.12"
 enum-map = "2.7.3"

crates/corrosion-tests/src/lib.rs

@@ -15,6 +15,7 @@ pub struct Trip {
 
 impl Trip {
     #[inline]
+    #[allow(clippy::new_without_default)]
     pub fn new() -> Self {
         let (tripwire, worker, tx) = tripwire::Tripwire::new_simple();
         Self {
@@ -50,7 +51,7 @@ pub struct TestSubsDb {
     matcher_conns: std::collections::BTreeMap<uuid::Uuid, types::sqlite::CrConn>,
     db_version: usize,
     pub trip: Trip,
-    btx: BroadcastingTransactor,
+    pub btx: BroadcastingTransactor,
 }
 
 impl TestSubsDb {
@@ -211,7 +212,7 @@ impl TestSubsDb {
             .changes_tx()
             .send(candidates)
             .await
-            .expect("failed to send changes")
+            .expect("failed to send changes");
     }
 
     #[inline]
@@ -348,7 +349,7 @@ pub fn query_to_string(
         statement
             .column_names()
             .into_iter()
-            .map(|name| pt::Cell::new(name))
+            .map(pt::Cell::new)
             .collect(),
     ));
 

crates/corrosion-tests/tests/db.rs

@@ -24,10 +24,11 @@ async fn read_server_row(id: usize, sp: &SplitPool) -> ServerRow {
         "SELECT endpoint,icao,tokens FROM servers WHERE rowid = ?",
         [id],
         |row| {
-            let mut v = Vec::with_capacity(3);
-            v.push(row.get::<_, SqliteValue>(0).unwrap());
-            v.push(row.get::<_, SqliteValue>(1).unwrap());
-            v.push(row.get::<_, SqliteValue>(2).unwrap());
+            let v = vec![
+                row.get::<_, SqliteValue>(0).unwrap(),
+                row.get::<_, SqliteValue>(1).unwrap(),
+                row.get::<_, SqliteValue>(2).unwrap(),
+            ];
             Ok(ServerRow::from_sql(&v).unwrap())
         },
     )

crates/corrosion-tests/tests/handshake.rs

@@ -1,24 +1,52 @@
-use corrosion::persistent::*;
+use corrosion::persistent::proto::{self, VersionedBuf, VersionedRequest, VersionedResponse};
 use quilkin_types::IcaoCode;
 
+/// Validates that v1 requests and responses can be serialized and deserialized
 #[test]
-fn version1_handshake() {
+fn version1_proto() {
+    use proto::v1;
+
     let icao = IcaoCode::new_testing([b'H'; 4]);
 
-    let chs = ClientHandshakeRequestV1 {
-        qcmp_port: 8998,
-        icao,
-    }
-    .write();
+    {
+        let req = v1::Request::Mutate(v1::MutateRequest {
+            qcmp_port: 8998,
+            icao,
+        })
+        .write()
+        .expect("failed to write mutate request");
 
-    let (version, chs) = ClientHandshake::read(1, &chs).unwrap();
+        let vb = VersionedBuf::try_parse(&req[2..])
+            .expect("failed to deserialize versioned buf for mutate request");
+        assert_eq!(vb.version, 1);
 
-    assert_eq!(version, 1);
-    assert_eq!(chs.client_details(), (8998, icao));
+        let req = vb
+            .deserialize_request()
+            .expect("failed to deserialize mutate request");
+        #[allow(unused_variables)]
+        {
+            assert!(matches!(
+                req,
+                proto::Request::V1(v1::Request::Mutate(v1::MutateRequest {
+                    qcmp_port: 8998,
+                    icao,
+                }))
+            ));
+        }
 
-    let shs = ServerHandshakeResponseV1 { accept: true }.write();
+        let res = v1::Response::Ok(v1::OkResponse::Mutate(v1::MutateResponse))
+            .write()
+            .expect("failed to serialize mutate response");
+        let vb = VersionedBuf::try_parse(&res[2..])
+            .expect("failed to deserialize versioned buf for mutate response");
+        assert_eq!(vb.version, 1);
 
-    let shs = ServerHandshake::read(1, &shs).unwrap();
-    let ServerHandshake::V1(v1) = shs;
-    assert!(v1.accept);
+        let res = vb
+            .deserialize_response()
+            .expect("failed to deserialize mutate response");
+        assert!(matches!(
+            res,
+            proto::Response::V1(v1::Response::Ok(v1::OkResponse::Mutate(v1::MutateResponse)))
+        ));
+    }
 }