Made R701 shareable via an internal Mutex, allowing get_name to be called while streaming records

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Cargo.toml

@@ -9,7 +9,7 @@ repository = "https://git.nicolabelluti.me/nicolabelluti/r701"
 async-stream = "0.3.6"
 chrono = { version = "0.4.38", default-features = false, features = ["clock"] }
 serde = { version = "1.0.203", default-features = false, features = ["derive"],  optional = true }
-tokio = { version = "1.53.1", default-features = false, features = ["net", "io-util"] }
+tokio = { version = "1.53.1", default-features = false, features = ["net", "io-util", "sync"] }
 tokio-stream = { version = "0.1.19", default-features = false }
 
 [features]

README.md

@@ -38,18 +38,13 @@ If you want to read about this reverse engineering attempt, check out
 
    #[tokio::main]
    async fn main() {
-       let mut r701 = R701::connect("127.0.0.1:5005").await.unwrap();
-
-       let records = r701
-           .into_record_iter()
-           .await
-           .unwrap()
-           .collect::<Vec<_>>()
-           .await;
+       let r701 = R701::connect("127.0.0.1:5005").await.unwrap();
+       let mut record_stream = r701.into_record_iter().await.unwrap();
+       let mut id = 0;
 
        println!("No\tMchn\tEnNo\t\tName\t\tMode\tIOMd\tDateTime\t");
-       for (id, record) in records.iter().rev().enumerate() {
-           let record = record.as_ref().unwrap();
+       while let Some(record) = record_stream.next().await {
+           let record = record.unwrap();
            let name = r701
                .get_name(record.employee_id)
                .await
@@ -66,6 +61,8 @@ If you want to read about this reverse engineering attempt, check out
                record.clock as u8,
                record.datetime.format("%Y/%m/%d  %H:%M:%S"),
            );
+
+           id += 1;
        }
    }
    ```

src/r701.rs

@@ -2,20 +2,28 @@ use crate::RecordIterator;
 use std::io::{Error, ErrorKind::InvalidData, Result};
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{TcpStream, ToSocketAddrs};
+use tokio::sync::Mutex;
+
+#[derive(Debug)]
+struct ConnectionState {
+    tcp_stream: TcpStream,
+    sequence_number: u16,
+}
 
 #[derive(Debug)]
 pub struct R701 {
-    tcp_stream: TcpStream,
-    sequence_number: u16,
+    state: Mutex<ConnectionState>,
 }
 
 impl R701 {
     #[allow(clippy::missing_errors_doc)]
     pub async fn connect(connection_info: impl ToSocketAddrs + Send) -> Result<Self> {
         // Create a new R701 struct
-        let mut new = Self {
-            tcp_stream: TcpStream::connect(connection_info).await?,
-            sequence_number: 0,
+        let new = Self {
+            state: Mutex::new(ConnectionState {
+                tcp_stream: TcpStream::connect(connection_info).await?,
+                sequence_number: 0,
+            }),
         };
 
         // Try to ping the endpoint
@@ -24,7 +32,12 @@ impl R701 {
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub async fn request(&mut self, payload: &[u8; 12]) -> Result<Vec<u8>> {
+    // The lock must stay held for the whole write+read round-trip, so that
+    // concurrent callers can't interleave their requests on the wire
+    #[allow(clippy::significant_drop_tightening)]
+    pub async fn request(&self, payload: &[u8; 12]) -> Result<Vec<u8>> {
+        let mut state = self.state.lock().await;
+
         // Create a blank request
         let mut request = [0x55, 0xaa, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0];
 
@@ -32,19 +45,19 @@ impl R701 {
         request[2..14].clone_from_slice(payload);
 
         // Insert the sequence number
-        request[14..].clone_from_slice(&self.sequence_number.to_le_bytes());
-        self.sequence_number += 1;
+        request[14..].clone_from_slice(&state.sequence_number.to_le_bytes());
+        state.sequence_number += 1;
 
         // Send the request
-        self.tcp_stream.write_all(&request).await?;
+        state.tcp_stream.write_all(&request).await?;
 
         // Create a buffer and return the response
-        let mut buffer = BufReader::new(&mut self.tcp_stream);
+        let mut buffer = BufReader::new(&mut state.tcp_stream);
         Ok(buffer.fill_buf().await?.to_vec())
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub async fn ping(&mut self) -> Result<()> {
+    pub async fn ping(&self) -> Result<()> {
         // Create a request with a payload of `01 80 00 00 00 00 00 00 00 00 00
         // 00` and get the response
         let response = self
@@ -61,7 +74,7 @@ impl R701 {
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub async fn get_name(&mut self, id: u32) -> Result<Option<String>> {
+    pub async fn get_name(&self, id: u32) -> Result<Option<String>> {
         // Create a request with a payload of `01 c7 xx xx xx xx 00 00 00 00 14
         // 00`, where `xx xx xx xx` is the employee ID in little endian
         let mut request = [0x01, 0xc7, 0, 0, 0, 0, 0, 0, 0, 0, 0x14, 0];
@@ -97,7 +110,7 @@ impl R701 {
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub async fn get_total_record_count(&mut self) -> Result<u16> {
+    pub async fn get_total_record_count(&self) -> Result<u16> {
         // Create a request with a payload of `01 b4 08 00 00 00 00 00 ff ff 00
         // 00` and get the response
         let response = self
@@ -121,7 +134,7 @@ impl R701 {
 
     #[allow(clippy::missing_errors_doc)]
     pub async fn get_record_bytes(
-        &mut self,
+        &self,
         total_records: u16,
         sequence_number: u16,
     ) -> Result<Vec<u8>> {
@@ -158,7 +171,7 @@ impl R701 {
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub async fn into_record_iter(&mut self) -> Result<RecordIterator<'_>> {
+    pub async fn into_record_iter(&self) -> Result<RecordIterator<'_>> {
         RecordIterator::from(self).await
     }
 }

src/record_iterator.rs

@@ -10,15 +10,16 @@ pub struct RecordIterator<'a> {
 
 impl<'a> RecordIterator<'a> {
     #[allow(clippy::missing_errors_doc)]
-    pub async fn from(r701: &'a mut R701) -> Result<Self> {
+    pub async fn from(r701: &'a R701) -> Result<Self> {
         // Get the total number of records
         let total_records = r701.get_total_record_count().await?;
 
         // Calculate the sequence number on which the last record resides and
-        // the index of the first `ff` byte, avoiding overflows
-        //
-        // TODO: Find a better way to do the multiplication and the division
-        // avoiding overflows
+        // the index of the first `ff` byte. The multiplication is done in
+        // u32 to avoid overflowing total_records * 12 (up to 786_420); the
+        // division result (at most 786_420 / 1024 = 768) always fits back
+        // into a u16, so the truncating cast below can never actually lose
+        // data
         #[allow(clippy::cast_possible_truncation)]
         let mut sequence_number = (u32::from(total_records) * 12 / 1024) as u16;
         let first_useless_byte_index = total_records as usize * 12 % 1024;