Made the library async using tokio and the iterator return a Result

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

Cargo.lock

@@ -1,6 +1,6 @@
 # This file is automatically @generated by Cargo.
 # It is not intended for manual editing.
-version = 3
+version = 4
 
 [[package]]
 name = "android-tzdata"
@@ -17,6 +17,28 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "async-stream"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476"
+dependencies = [
+ "async-stream-impl",
+ "futures-core",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "async-stream-impl"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "autocfg"
 version = "1.3.0"
@@ -29,6 +51,12 @@ version = "3.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "79296716171880943b8470b5f8d03aa55eb2e645a4874bdbb28adb49162e012c"
 
+[[package]]
+name = "bytes"
+version = "1.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
+
 [[package]]
 name = "cc"
 version = "1.0.99"
@@ -60,6 +88,12 @@ version = "0.8.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "06ea2b9bc92be3c2baa9334a323ebca2d6f074ff852cd1d7b11064035cd3868f"
 
+[[package]]
+name = "futures-core"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e"
+
 [[package]]
 name = "iana-time-zone"
 version = "0.1.60"
@@ -94,9 +128,9 @@ dependencies = [
 
 [[package]]
 name = "libc"
-version = "0.2.155"
+version = "0.2.189"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "97b3888a4aecf77e811145cadf6eef5901f4782c53886191b2f693f24761847c"
+checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
 
 [[package]]
 name = "log"
@@ -104,6 +138,17 @@ version = "0.4.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "90ed8c1e510134f979dbc4f070f87d4313098b704861a105fe34231c70a3901c"
 
+[[package]]
+name = "mio"
+version = "1.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4b18443e9c262bfe8fa82f51666e2642c53393f7e5c27b3e1aeab922cff5b9d8"
+dependencies = [
+ "libc",
+ "wasi",
+ "windows-sys",
+]
+
 [[package]]
 name = "num-traits"
 version = "0.2.19"
@@ -119,6 +164,12 @@ version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3fdb12b2476b595f9358c5161aa467c2438859caa136dec86c26fdd2efe17b92"
 
+[[package]]
+name = "pin-project-lite"
+version = "0.2.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
+
 [[package]]
 name = "proc-macro2"
 version = "1.0.85"
@@ -141,8 +192,11 @@ dependencies = [
 name = "r701"
 version = "0.1.0"
 dependencies = [
+ "async-stream",
  "chrono",
  "serde",
+ "tokio",
+ "tokio-stream",
 ]
 
 [[package]]
@@ -165,6 +219,16 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "socket2"
+version = "0.6.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4"
+dependencies = [
+ "libc",
+ "windows-sys",
+]
+
 [[package]]
 name = "syn"
 version = "2.0.66"
@@ -176,12 +240,43 @@ dependencies = [
  "unicode-ident",
 ]
 
+[[package]]
+name = "tokio"
+version = "1.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
+dependencies = [
+ "bytes",
+ "libc",
+ "mio",
+ "pin-project-lite",
+ "socket2",
+ "windows-sys",
+]
+
+[[package]]
+name = "tokio-stream"
+version = "0.1.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b"
+dependencies = [
+ "futures-core",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3354b9ac3fae1ff6755cb6db53683adb661634f67557942dea4facebec0fee4b"
 
+[[package]]
+name = "wasi"
+version = "0.11.1+wasi-snapshot-preview1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
+
 [[package]]
 name = "wasm-bindgen"
 version = "0.2.92"
@@ -245,6 +340,21 @@ dependencies = [
  "windows-targets",
 ]
 
+[[package]]
+name = "windows-link"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
+
+[[package]]
+name = "windows-sys"
+version = "0.61.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
+dependencies = [
+ "windows-link",
+]
+
 [[package]]
 name = "windows-targets"
 version = "0.52.5"

Cargo.toml

@@ -6,8 +6,11 @@ license = "GNU AGPLv3.0"
 repository = "https://git.nicolabelluti.me/nicolabelluti/r701"
 
 [dependencies]
+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-stream = { version = "0.1.19", default-features = false }
 
 [features]
 serde = ["chrono/serde", "dep:serde"]

README.md

@@ -28,35 +28,44 @@ If you want to read about this reverse engineering attempt, check out
 
 2. Use the library in you project:
 
+   This library is async and built on [tokio](https://tokio.rs); add it to your
+   own `[dependencies]` (e.g. `tokio = { version = "1", features =
+   ["rt-multi-thread", "macros"] }`) to run the example below.
+
    ```rust
    use r701::R701;
+   use tokio_stream::StreamExt;
 
-   fn main() {
+   #[tokio::main]
-       let mut r701 = R701::connect("127.0.0.1:5005").unwrap();
+   async fn main() {
+       let mut r701 = R701::connect("127.0.0.1:5005").await.unwrap();
 
-       println!("No\tMchn\tEnNo\t\tName\t\tMode\tIOMd\tDateTime\t");
+       let records = r701
-       r701.into_record_iter()
+           .into_record_iter()
+           .await
            .unwrap()
            .collect::<Vec<_>>()
-           .iter()
+           .await;
-           .rev()
-           .enumerate()
-           .for_each(|(id, record)| {
-               let name = r701
-                   .get_name(record.employee_id)
-                   .unwrap()
-                   .unwrap_or(format!("user #{}", record.employee_id));
 
-               println!(
+       println!("No\tMchn\tEnNo\t\tName\t\tMode\tIOMd\tDateTime\t");
-                   "{:0>6}\t{}\t{:0>9}\t{: <10}\t{}\t{}\t{}",
+       for (id, record) in records.iter().rev().enumerate() {
-                   id + 1,
+           let record = record.as_ref().unwrap();
-                   1,
+           let name = r701
-                   record.employee_id,
+               .get_name(record.employee_id)
-                   name,
+               .await
-                   35,
+               .unwrap()
-                   record.clock as u8,
+               .unwrap_or(format!("user #{}", record.employee_id));
-                   record.datetime.format("%Y/%m/%d  %H:%M:%S"),
+
-               );
+           println!(
-           });
+               "{:0>6}\t{}\t{:0>9}\t{: <10}\t{}\t{}\t{}",
+               id + 1,
+               1,
+               record.employee_id,
+               name,
+               35,
+               record.clock as u8,
+               record.datetime.format("%Y/%m/%d  %H:%M:%S"),
+           );
+       }
    }
    ```

src/r701.rs

@@ -1,6 +1,7 @@
 use crate::RecordIterator;
-use std::io::{BufRead, BufReader, Error, ErrorKind::InvalidData, Result, Write};
+use std::io::{Error, ErrorKind::InvalidData, Result};
-use std::net::{TcpStream, ToSocketAddrs};
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::net::{TcpStream, ToSocketAddrs};
 
 #[derive(Debug)]
 pub struct R701 {
@@ -10,20 +11,20 @@ pub struct R701 {
 
 impl R701 {
     #[allow(clippy::missing_errors_doc)]
-    pub fn connect(connection_info: impl ToSocketAddrs) -> Result<Self> {
+    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)?,
+            tcp_stream: TcpStream::connect(connection_info).await?,
             sequence_number: 0,
         };
 
         // Try to ping the endpoint
-        new.ping()?;
+        new.ping().await?;
         Ok(new)
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub fn request(&mut self, payload: &[u8; 12]) -> Result<Vec<u8>> {
+    pub async fn request(&mut self, payload: &[u8; 12]) -> Result<Vec<u8>> {
         // Create a blank request
         let mut request = [0x55, 0xaa, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0];
 
@@ -35,18 +36,20 @@ impl R701 {
         self.sequence_number += 1;
 
         // Send the request
-        self.tcp_stream.write_all(&request)?;
+        self.tcp_stream.write_all(&request).await?;
 
         // Create a buffer and return the response
-        let mut buffer = BufReader::new(&self.tcp_stream);
+        let mut buffer = BufReader::new(&mut self.tcp_stream);
-        Ok(buffer.fill_buf()?.to_vec())
+        Ok(buffer.fill_buf().await?.to_vec())
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub fn ping(&mut self) -> Result<()> {
+    pub async fn ping(&mut 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.request(&[0x01, 0x80, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0])?;
+        let response = self
+            .request(&[0x01, 0x80, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0])
+            .await?;
 
         // If the response is not `aa 55 01 01 00 00 00 00 00 00` then return an
         // error
@@ -58,14 +61,14 @@ impl R701 {
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub fn get_name(&mut self, id: u32) -> Result<Option<String>> {
+    pub async fn get_name(&mut 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];
         request[2..6].clone_from_slice(&id.to_le_bytes());
 
         // Get the response
-        let response = self.request(&request)?;
+        let response = self.request(&request).await?;
 
         // If the response length is right but the header is `01 00 00 00 00 00
         // 00 00` then the request is been succesful but the name was not found
@@ -94,10 +97,12 @@ impl R701 {
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub fn get_total_record_count(&mut self) -> Result<u16> {
+    pub async fn get_total_record_count(&mut 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.request(&[0x01, 0xb4, 0x08, 0, 0, 0, 0, 0, 0xff, 0xff, 0, 0])?;
+        let response = self
+            .request(&[0x01, 0xb4, 0x08, 0, 0, 0, 0, 0, 0xff, 0xff, 0, 0])
+            .await?;
 
         // If one between the response length or the response header is wrong
         // return an error
@@ -115,7 +120,7 @@ impl R701 {
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub fn get_record_bytes(
+    pub async fn get_record_bytes(
         &mut self,
         total_records: u16,
         sequence_number: u16,
@@ -137,7 +142,7 @@ impl R701 {
         }
 
         // Get the response
-        let response = self.request(&request)?;
+        let response = self.request(&request).await?;
 
         // If one between the response length, the response header or the last
         // two bits is wrong return an error
@@ -153,7 +158,7 @@ impl R701 {
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub fn into_record_iter(&mut self) -> Result<RecordIterator<'_>> {
+    pub async fn into_record_iter(&mut self) -> Result<RecordIterator<'_>> {
-        RecordIterator::from(self)
+        RecordIterator::from(self).await
     }
 }

src/record_iterator.rs

@@ -1,20 +1,18 @@
 use crate::{Record, R701};
-use std::io::Result;
+use std::io::{Error, ErrorKind::InvalidData, Result};
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio_stream::Stream;
 
-#[derive(Debug)]
 pub struct RecordIterator<'a> {
-    r701: &'a mut R701,
+    inner: Pin<Box<dyn Stream<Item = Result<Record>> + 'a>>,
-    input_buffer: Vec<u8>,
-    sequence_number: u16,
-    total_records: u16,
-    record_count: u16,
 }
 
 impl<'a> RecordIterator<'a> {
     #[allow(clippy::missing_errors_doc)]
-    pub fn from(r701: &'a mut R701) -> Result<Self> {
+    pub async fn from(r701: &'a mut R701) -> Result<Self> {
         // Get the total number of records
-        let total_records = r701.get_total_record_count()?;
+        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
@@ -22,66 +20,64 @@ impl<'a> RecordIterator<'a> {
         // TODO: Find a better way to do the multiplication and the division
         // avoiding overflows
         #[allow(clippy::cast_possible_truncation)]
-        let sequence_number = (u32::from(total_records) * 12 / 1024) as u16;
+        let mut sequence_number = (u32::from(total_records) * 12 / 1024) as u16;
         let first_useless_byte_index = total_records as usize * 12 % 1024;
 
         // The endpoint expects the first block of records to be sent first
-        r701.get_record_bytes(total_records, 0)?;
+        r701.get_record_bytes(total_records, 0).await?;
 
         // Get the last records and cut out all the trailing `ff` bytes
-        let input_buffer = r701
+        let mut input_buffer = r701
-            .get_record_bytes(total_records, sequence_number)?
+            .get_record_bytes(total_records, sequence_number)
-            .drain(..first_useless_byte_index)
+            .await?;
-            .collect::<Vec<u8>>();
+        input_buffer.truncate(first_useless_byte_index);
+
+        let mut record_count = total_records;
+
+        // Build the record stream, yielding an `Err` both on network errors and
+        // on malformed records, so the caller can fail
+        let inner = async_stream::try_stream! {
+            while record_count != 0 {
+                record_count -= 1;
+
+                // If the input buffer is almost empty, make another request to
+                // the endpoint asking for more data
+                if input_buffer.len() < 12 {
+                    // If the buffer is almost empty but the sequence number is
+                    // already zero, stop iterating
+                    if sequence_number == 0 {
+                        break;
+                    }
+
+                    sequence_number -= 1;
+
+                    // Request new bytes
+                    let bytes = r701.get_record_bytes(total_records, sequence_number).await?;
+
+                    // Put the bytes at the start of the vector
+                    input_buffer.splice(0..0, bytes);
+                }
+
+                // Yield a new Record
+                yield input_buffer
+                    .drain(input_buffer.len() - 12..)
+                    .as_slice()
+                    .try_into()
+                    .map_err(|error: &str| Error::new(InvalidData, error))?;
+            }
+        };
 
         // Return a new Iterator
         Ok(Self {
-            r701,
+            inner: Box::pin(inner),
-            input_buffer,
-            sequence_number,
-            total_records,
-            record_count: total_records,
         })
     }
 }
 
-impl Iterator for RecordIterator<'_> {
+impl Stream for RecordIterator<'_> {
-    type Item = Record;
+    type Item = Result<Record>;
 
-    fn next(&mut self) -> Option<Self::Item> {
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        // Stop iterating when there are no more records
+        self.inner.as_mut().poll_next(cx)
-        if self.record_count == 0 {
-            return None;
-        }
-
-        self.record_count -= 1;
-
-        // If the input buffer is almost empty, make another request to the
-        // endpoint asking for more data
-        if self.input_buffer.len() < 12 {
-            // If the buffer is almost empty but the sequence number is already
-            // zero, stop iterating
-            if self.sequence_number == 0 {
-                return None;
-            }
-
-            self.sequence_number -= 1;
-
-            // Request new bytes
-            let bytes = self
-                .r701
-                .get_record_bytes(self.total_records, self.sequence_number)
-                .ok()?;
-
-            // Put the bytes at the start of the vector
-            self.input_buffer.splice(0..0, bytes.iter().copied());
-        }
-
-        // Return a new Record
-        self.input_buffer
-            .drain(self.input_buffer.len() - 12..)
-            .as_slice()
-            .try_into()
-            .ok()
     }
 }