diff --git a/Cargo.lock b/Cargo.lock index 7d69cab..489ba95 100644 --- a/Cargo.lock +++ b/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" diff --git a/Cargo.toml b/Cargo.toml index f111bb8..48d7a5c 100644 --- a/Cargo.toml +++ b/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"] diff --git a/README.md b/README.md index 3e371db..c4ca038 100644 --- a/README.md +++ b/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() { - let mut r701 = R701::connect("127.0.0.1:5005").unwrap(); + #[tokio::main] + 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"); - r701.into_record_iter() + let records = r701 + .into_record_iter() + .await .unwrap() .collect::>() - .iter() - .rev() - .enumerate() - .for_each(|(id, record)| { - let name = r701 - .get_name(record.employee_id) - .unwrap() - .unwrap_or(format!("user #{}", record.employee_id)); + .await; - 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"), - ); - }); + 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(); + let name = r701 + .get_name(record.employee_id) + .await + .unwrap() + .unwrap_or(format!("user #{}", record.employee_id)); + + 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"), + ); + } } ``` diff --git a/src/r701.rs b/src/r701.rs index f212790..c9e0fec 100644 --- a/src/r701.rs +++ b/src/r701.rs @@ -1,6 +1,7 @@ use crate::RecordIterator; -use std::io::{BufRead, BufReader, Error, ErrorKind::InvalidData, Result, Write}; -use std::net::{TcpStream, ToSocketAddrs}; +use std::io::{Error, ErrorKind::InvalidData, Result}; +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 { + pub async fn connect(connection_info: impl ToSocketAddrs + Send) -> Result { // 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> { + pub async fn request(&mut self, payload: &[u8; 12]) -> Result> { // 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); - Ok(buffer.fill_buf()?.to_vec()) + let mut buffer = BufReader::new(&mut self.tcp_stream); + 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> { + pub async fn get_name(&mut self, id: u32) -> Result> { // 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 { + pub async fn get_total_record_count(&mut self) -> Result { // 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::from(self) + pub async fn into_record_iter(&mut self) -> Result> { + RecordIterator::from(self).await } } diff --git a/src/record_iterator.rs b/src/record_iterator.rs index 8f28439..0ea62f4 100644 --- a/src/record_iterator.rs +++ b/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, - input_buffer: Vec, - sequence_number: u16, - total_records: u16, - record_count: u16, + inner: Pin> + 'a>>, } impl<'a> RecordIterator<'a> { #[allow(clippy::missing_errors_doc)] - pub fn from(r701: &'a mut R701) -> Result { + pub async fn from(r701: &'a mut R701) -> Result { // 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 - .get_record_bytes(total_records, sequence_number)? - .drain(..first_useless_byte_index) - .collect::>(); + let mut input_buffer = r701 + .get_record_bytes(total_records, sequence_number) + .await?; + 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, - input_buffer, - sequence_number, - total_records, - record_count: total_records, + inner: Box::pin(inner), }) } } -impl Iterator for RecordIterator<'_> { - type Item = Record; +impl Stream for RecordIterator<'_> { + type Item = Result; - fn next(&mut self) -> Option { - // Stop iterating when there are no more records - 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() + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.inner.as_mut().poll_next(cx) } }