Test parsing keepalive messages

Author: simolus3

crates/core/src/bson/de.rs

@@ -140,13 +140,14 @@ impl<'de, 'a> de::Deserializer<'de> for &'a mut Deserializer<'de> {
                 let (_, bytes) = self.parser.read_binary()?;
                 visitor.visit_borrowed_bytes(bytes)
             }
-            ElementType::ObjectId => todo!(),
+            ElementType::ObjectId => visitor.visit_borrowed_bytes(self.parser.read_object_id()?),
             ElementType::Boolean => visitor.visit_bool(self.parser.read_bool()?),
-            ElementType::DatetimeUtc => todo!(),
+            ElementType::DatetimeUtc | ElementType::Timestamp => {
+                visitor.visit_u64(self.parser.read_uint64()?)
+            }
             ElementType::Null | ElementType::Undefined => visitor.visit_unit(),
             ElementType::Int32 => visitor.visit_i32(self.parser.read_int32()?),
             ElementType::Int64 => visitor.visit_i64(self.parser.read_int64()?),
-            ElementType::Timestamp => todo!(),
         }
     }
 
@@ -196,9 +197,21 @@ impl<'de, 'a> de::Deserializer<'de> for &'a mut Deserializer<'de> {
         }
     }
 
+    fn deserialize_newtype_struct<V>(
+        self,
+        name: &'static str,
+        visitor: V,
+    ) -> Result<V::Value, Self::Error>
+    where
+        V: Visitor<'de>,
+    {
+        self.prepare_to_read_value()?;
+        visitor.visit_newtype_struct(self)
+    }
+
     forward_to_deserialize_any! {
         bool i8 i16 i32 i64 i128 u8 u16 u32 u64 u128 f32 f64 char str string
-        bytes byte_buf unit unit_struct newtype_struct seq tuple
+        bytes byte_buf unit unit_struct  seq tuple
         tuple_struct map struct ignored_any identifier
     }
 }

crates/core/src/bson/mod.rs

@@ -15,7 +15,9 @@ pub fn from_bytes<'de, T: Deserialize<'de>>(bytes: &'de [u8]) -> Result<T, BsonE
 
 #[cfg(test)]
 mod test {
-    use crate::sync::line::SyncLine;
+    use core::assert_matches::assert_matches;
+
+    use crate::sync::line::{SyncLine, TokenExpiresIn};
 
     use super::*;
 
@@ -44,4 +46,12 @@ mod test {
 
         assert_eq!(checkpoint.buckets.len(), 1);
     }
+
+    #[test]
+    fn test_newtype_tuple() {
+        let bson = b"\x1b\x00\x00\x00\x10token_expires_in\x00<\x00\x00\x00\x00";
+
+        let expected: SyncLine = from_bytes(bson.as_slice()).expect("should deserialize");
+        assert_matches!(expected, SyncLine::KeepAlive(TokenExpiresIn(60)));
+    }
 }

crates/core/src/bson/parser.rs

@@ -1,6 +1,7 @@
 use core::ffi::CStr;
 
 use super::{error::ErrorKind, BsonError};
+use num_traits::{FromBytes, Num};
 
 pub struct Parser<'de> {
     offset: usize,
@@ -49,6 +50,11 @@ impl<'de> Parser<'de> {
         Ok(value)
     }
 
+    fn advance_bytes<const N: usize>(&mut self) -> Result<&'de [u8; N], BsonError> {
+        let bytes = self.advance_checked(N)?;
+        Ok(bytes.try_into().expect("should have correct length"))
+    }
+
     pub fn read_cstr(&mut self) -> Result<&'de str, BsonError> {
         let raw = CStr::from_bytes_until_nul(self.remaining_input)
             .map_err(|_| self.error(ErrorKind::UnterminatedCString))?;
@@ -60,11 +66,15 @@ impl<'de> Parser<'de> {
         Ok(str)
     }
 
+    fn read_number<const N: usize, T: Num + FromBytes<Bytes = [u8; N]>>(
+        &mut self,
+    ) -> Result<T, BsonError> {
+        let bytes = self.advance_bytes::<N>()?;
+        Ok(T::from_le_bytes(&bytes))
+    }
+
     pub fn read_int32(&mut self) -> Result<i32, BsonError> {
-        let slice = self.advance_checked(4)?;
-        Ok(i32::from_le_bytes(
-            slice.try_into().expect("should have correct length"),
-        ))
+        self.read_number()
     }
 
     fn read_length(&mut self) -> Result<usize, BsonError> {
@@ -75,24 +85,26 @@ impl<'de> Parser<'de> {
     }
 
     pub fn read_int64(&mut self) -> Result<i64, BsonError> {
-        let slice = self.advance_checked(8)?;
-        Ok(i64::from_le_bytes(
-            slice.try_into().expect("should have correct length"),
-        ))
+        self.read_number()
+    }
+
+    pub fn read_uint64(&mut self) -> Result<u64, BsonError> {
+        self.read_number()
     }
 
     pub fn read_double(&mut self) -> Result<f64, BsonError> {
-        let slice = self.advance_checked(8)?;
-        Ok(f64::from_le_bytes(
-            slice.try_into().expect("should have correct length"),
-        ))
+        self.read_number()
     }
 
     pub fn read_bool(&mut self) -> Result<bool, BsonError> {
         let byte = self.advance_byte()?;
         Ok(byte != 0)
     }
 
+    pub fn read_object_id(&mut self) -> Result<&'de [u8], BsonError> {
+        self.advance_checked(12)
+    }
+
     /// Reads a BSON string, `string ::= int32 (byte*) unsigned_byte(0)`
     pub fn read_string(&mut self) -> Result<&'de str, BsonError> {
         let length_including_null = self.read_length()?;
@@ -170,7 +182,7 @@ impl<'de> Parser<'de> {
         Ok(self.subreader(parsed_size)?.remaining())
     }
 
-    /// If only a single byte is left in the current scope, assert that it is a zero byte.
+    /// If only a single byte is left in the current scope, validate that it is a zero byte.
     ///
     /// Otherwise returns false as we haven't reached the end of a document.
     pub fn end_document(&mut self) -> Result<bool, BsonError> {

crates/core/src/sync/line.rs

@@ -388,7 +388,7 @@ mod tests {
 
     #[test]
     fn parse_checkpoint_diff_no_write_checkpoint() {
-        let SyncLine::CheckpointDiff(diff) = deserialize(
+        let SyncLine::CheckpointDiff(_diff) = deserialize(
             r#"{"checkpoint_diff":{"last_op_id":"12","updated_buckets":[{"bucket":"a","count":12,"checksum":0,"priority":3}],"removed_buckets":[]}}"#,
         ) else {
             panic!("Expected checkpoint diff")

dart/test/goldens/simple_iteration.json

@@ -65,6 +65,13 @@
       }
     ]
   },
+  {
+    "operation": "line_text",
+    "data": {
+      "token_expires_in": 60
+    },
+    "output": []
+  },
   {
     "operation": "line_text",
     "data": {
@@ -146,5 +153,18 @@
         }
       }
     ]
+  },
+  {
+    "operation": "line_text",
+    "data": {
+      "token_expires_in": 10
+    },
+    "output": [
+      {
+        "FetchCredentials": {
+          "did_expire": false
+        }
+      }
+    ]
   }
 ]

dart/test/sync_test.dart

@@ -162,11 +162,14 @@ void _syncTests<T>({
           ],
         },
       });
+      syncLine({'token_expires_in': 60});
       pushSyncData('a', '1', '1', 'PUT', {'col': 'hi'});
 
       syncLine({
         'checkpoint_complete': {'last_op_id': '1'},
       });
+
+      syncLine({'token_expires_in': 10});
     });
   });