More WIP

Signed-off-by: Martin Tzvetanov Grigorov <[email protected]>

Author: martin-g

avro/src/bigdecimal.rs

@@ -22,89 +22,113 @@
   pub mod sync {
     sync!();
     replace!(
-      tokio::io::AsyncRead => std::io::Read,
+      bigdecimal::tokio => bigdecimal::sync,
       decode::tokio => decode::sync,
+      encode::tokio => encode::sync,
+      error::tokio => error::sync,
+      schema::tokio => schema::sync,
+      util::tokio => util::sync,
       #[tokio::test] => #[test]
     );
   }
 )]
-mod bigdecimal {}
-use crate::{
-    AvroResult,
-    decode::tokio::{decode_len, decode_long},
-    encode::{encode_bytes, encode_long},
-    error::Details,
-    types::Value,
-};
-pub use bigdecimal::BigDecimal;
-use num_bigint::BigInt;
-use std::io::Read;
-
-pub(crate) fn big_decimal_as_bytes(decimal: &BigDecimal) -> AvroResult<Vec<u8>> {
-    let mut buffer: Vec<u8> = Vec::new();
-    let (big_int, exponent): (BigInt, i64) = decimal.as_bigint_and_exponent();
-    let big_endian_value: Vec<u8> = big_int.to_signed_bytes_be();
-    encode_bytes(&big_endian_value, &mut buffer)?;
-    encode_long(exponent, &mut buffer)?;
-
-    Ok(buffer)
-}
-
-pub(crate) fn serialize_big_decimal(decimal: &BigDecimal) -> AvroResult<Vec<u8>> {
-    // encode big decimal, without global size
-    let buffer = big_decimal_as_bytes(decimal)?;
+mod bigdecimal {
+    use crate::{
+        AvroResult,
+        decode::tokio::{decode_len, decode_long},
+        encode::tokio::{encode_bytes, encode_long},
+        error::tokio::Details,
+        types::tokio::Value,
+    };
+    pub use bigdecimal::BigDecimal;
+    use num_bigint::BigInt;
+    use std::io::Read;
+
+    pub(crate) fn big_decimal_as_bytes(decimal: &BigDecimal) -> AvroResult<Vec<u8>> {
+        let mut buffer: Vec<u8> = Vec::new();
+        let (big_int, exponent): (BigInt, i64) = decimal.as_bigint_and_exponent();
+        let big_endian_value: Vec<u8> = big_int.to_signed_bytes_be();
+        encode_bytes(&big_endian_value, &mut buffer)?;
+        encode_long(exponent, &mut buffer)?;
+
+        Ok(buffer)
+    }
 
-    // encode global size and content
-    let mut final_buffer: Vec<u8> = Vec::new();
-    encode_bytes(&buffer, &mut final_buffer)?;
+    pub(crate) fn serialize_big_decimal(decimal: &BigDecimal) -> AvroResult<Vec<u8>> {
+        // encode big decimal, without global size
+        let buffer = big_decimal_as_bytes(decimal)?;
 
-    Ok(final_buffer)
-}
+        // encode global size and content
+        let mut final_buffer: Vec<u8> = Vec::new();
+        encode_bytes(&buffer, &mut final_buffer)?;
 
-pub(crate) async fn deserialize_big_decimal(bytes: &Vec<u8>) -> AvroResult<BigDecimal> {
-    let mut bytes: &[u8] = bytes.as_slice();
-    let mut big_decimal_buffer = match decode_len(&mut bytes).await {
-        Ok(size) => vec![0u8; size],
-        Err(err) => return Err(Details::BigDecimalLen(Box::new(err)).into()),
-    };
-
-    bytes
-        .read_exact(&mut big_decimal_buffer[..])
-        .map_err(Details::ReadDouble)?;
+        Ok(final_buffer)
+    }
 
-    match decode_long(&mut bytes).await {
-        Ok(Value::Long(scale_value)) => {
-            let big_int: BigInt = BigInt::from_signed_bytes_be(&big_decimal_buffer);
-            let decimal = BigDecimal::new(big_int, scale_value);
-            Ok(decimal)
+    pub(crate) async fn deserialize_big_decimal(bytes: &Vec<u8>) -> AvroResult<BigDecimal> {
+        let mut bytes: &[u8] = bytes.as_slice();
+        let mut big_decimal_buffer = match decode_len(&mut bytes).await {
+            Ok(size) => vec![0u8; size],
+            Err(err) => return Err(Details::BigDecimalLen(Box::new(err)).into()),
+        };
+
+        bytes
+            .read_exact(&mut big_decimal_buffer[..])
+            .map_err(Details::ReadDouble)?;
+
+        match decode_long(&mut bytes).await {
+            Ok(Value::Long(scale_value)) => {
+                let big_int: BigInt = BigInt::from_signed_bytes_be(&big_decimal_buffer);
+                let decimal = BigDecimal::new(big_int, scale_value);
+                Ok(decimal)
+            }
+            _ => Err(Details::BigDecimalScale.into()),
         }
-        _ => Err(Details::BigDecimalScale.into()),
     }
-}
 
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::{Codec, Reader, Schema, Writer, error::Error, types::Record};
-    use apache_avro_test_helper::TestResult;
-    use bigdecimal::{One, Zero};
-    use pretty_assertions::assert_eq;
-    use std::{
-        fs::File,
-        io::BufReader,
-        ops::{Div, Mul},
-        str::FromStr,
-    };
-
-    #[tokio::test]
-    async fn test_avro_3779_bigdecimal_serial() -> TestResult {
-        let value: BigDecimal =
-            bigdecimal::BigDecimal::from(-1421).div(bigdecimal::BigDecimal::from(2));
-        let mut current: BigDecimal = BigDecimal::one();
-
-        for iter in 1..180 {
-            let buffer: Vec<u8> = serialize_big_decimal(&current)?;
+    #[cfg(test)]
+    mod tests {
+        use super::*;
+        use crate::{
+            codec::tokio::Codec, error::tokio::Error, reader::tokio::Reader, schema::tokio::Schema,
+            types::tokio::Record, writer::tokio::Writer,
+        };
+        use apache_avro_test_helper::TestResult;
+        use bigdecimal::{One, Zero};
+        use pretty_assertions::assert_eq;
+        use std::{
+            fs::File,
+            io::BufReader,
+            ops::{Div, Mul},
+            str::FromStr,
+        };
+
+        #[tokio::test]
+        async fn test_avro_3779_bigdecimal_serial() -> TestResult {
+            let value: BigDecimal =
+                bigdecimal::BigDecimal::from(-1421).div(bigdecimal::BigDecimal::from(2));
+            let mut current: BigDecimal = BigDecimal::one();
+
+            for iter in 1..180 {
+                let buffer: Vec<u8> = serialize_big_decimal(&current)?;
+
+                let mut as_slice = buffer.as_slice();
+                decode_long(&mut as_slice).await?;
+
+                let mut result: Vec<u8> = Vec::new();
+                result.extend_from_slice(as_slice);
+
+                let deserialize_big_decimal: Result<BigDecimal, Error> =
+                    deserialize_big_decimal(&result);
+                assert!(
+                    deserialize_big_decimal.is_ok(),
+                    "can't deserialize for iter {iter}"
+                );
+                assert_eq!(current, deserialize_big_decimal?, "not equals for {iter}");
+                current = current.mul(&value);
+            }
 
+            let buffer: Vec<u8> = serialize_big_decimal(&BigDecimal::zero())?;
             let mut as_slice = buffer.as_slice();
             decode_long(&mut as_slice).await?;
 
@@ -115,36 +139,20 @@ mod tests {
                 deserialize_big_decimal(&result);
             assert!(
                 deserialize_big_decimal.is_ok(),
-                "can't deserialize for iter {iter}"
+                "can't deserialize for zero"
+            );
+            assert_eq!(
+                BigDecimal::zero(),
+                deserialize_big_decimal?,
+                "not equals for zero"
             );
-            assert_eq!(current, deserialize_big_decimal?, "not equals for {iter}");
-            current = current.mul(&value);
-        }
 
-        let buffer: Vec<u8> = serialize_big_decimal(&BigDecimal::zero())?;
-        let mut as_slice = buffer.as_slice();
-        decode_long(&mut as_slice).await?;
-
-        let mut result: Vec<u8> = Vec::new();
-        result.extend_from_slice(as_slice);
-
-        let deserialize_big_decimal: Result<BigDecimal, Error> = deserialize_big_decimal(&result);
-        assert!(
-            deserialize_big_decimal.is_ok(),
-            "can't deserialize for zero"
-        );
-        assert_eq!(
-            BigDecimal::zero(),
-            deserialize_big_decimal?,
-            "not equals for zero"
-        );
-
-        Ok(())
-    }
+            Ok(())
+        }
 
-    #[tokio::test]
-    async fn test_avro_3779_record_with_bg() -> TestResult {
-        let schema_str = r#"
+        #[tokio::test]
+        async fn test_avro_3779_record_with_bg() -> TestResult {
+            let schema_str = r#"
         {
           "type": "record",
           "name": "test",
@@ -157,66 +165,67 @@ mod tests {
           ]
         }
         "#;
-        let schema = Schema::parse_str(schema_str)?;
-
-        // build record with big decimal value
-        let mut record = Record::new(&schema).unwrap();
-        let val = BigDecimal::new(BigInt::from(12), 2);
-        record.put("field_name", val.clone());
-
-        // write a record
-        let codec = Codec::Null;
-        let mut writer = Writer::builder()
-            .schema(&schema)
-            .codec(codec)
-            .writer(Vec::new())
-            .build();
-
-        writer.append(record.clone())?;
-        writer.flush()?;
-
-        // read record
-        let wrote_data = writer.into_inner()?;
-        let mut reader = Reader::new(&wrote_data[..])?;
-
-        let value = reader.next().await.unwrap()?;
-
-        // extract field value
-        let big_decimal_value: &Value = match value {
-            Value::Record(ref fields) => Ok(&fields[0].1),
-            other => Err(format!("Expected a Value::Record, got: {other:?}")),
-        }?;
-
-        let x1res: &BigDecimal = match big_decimal_value {
-            Value::BigDecimal(s) => Ok(s),
-            other => Err(format!("Expected Value::BigDecimal, got: {other:?}")),
-        }?;
-        assert_eq!(&val, x1res);
-
-        Ok(())
-    }
+            let schema = Schema::parse_str(schema_str)?;
+
+            // build record with big decimal value
+            let mut record = Record::new(&schema).unwrap();
+            let val = BigDecimal::new(BigInt::from(12), 2);
+            record.put("field_name", val.clone());
+
+            // write a record
+            let codec = Codec::Null;
+            let mut writer = Writer::builder()
+                .schema(&schema)
+                .codec(codec)
+                .writer(Vec::new())
+                .build();
+
+            writer.append(record.clone())?;
+            writer.flush()?;
+
+            // read record
+            let wrote_data = writer.into_inner()?;
+            let mut reader = Reader::new(&wrote_data[..])?;
+
+            let value = reader.next().await.unwrap()?;
+
+            // extract field value
+            let big_decimal_value: &Value = match value {
+                Value::Record(ref fields) => Ok(&fields[0].1),
+                other => Err(format!("Expected a Value::Record, got: {other:?}")),
+            }?;
+
+            let x1res: &BigDecimal = match big_decimal_value {
+                Value::BigDecimal(s) => Ok(s),
+                other => Err(format!("Expected Value::BigDecimal, got: {other:?}")),
+            }?;
+            assert_eq!(&val, x1res);
+
+            Ok(())
+        }
 
-    #[tokio::test]
-    async fn test_avro_3779_from_java_file() -> TestResult {
-        // Open file generated with Java code to ensure compatibility
-        // with Java big decimal logical type.
-        let file: File = File::open("./tests/bigdec.avro")?;
-        let mut reader = Reader::new(BufReader::new(&file))?;
-        let next_element = reader.next().await;
-        assert!(next_element.is_some());
-        let value = next_element.unwrap()?;
-        let bg = match value {
-            Value::Record(ref fields) => Ok(&fields[0].1),
-            other => Err(format!("Expected a Value::Record, got: {other:?}")),
-        }?;
-        let value_big_decimal = match bg {
-            Value::BigDecimal(val) => Ok(val),
-            other => Err(format!("Expected a Value::BigDecimal, got: {other:?}")),
-        }?;
-
-        let ref_value = BigDecimal::from_str("2.24")?;
-        assert_eq!(&ref_value, value_big_decimal);
-
-        Ok(())
+        #[tokio::test]
+        async fn test_avro_3779_from_java_file() -> TestResult {
+            // Open file generated with Java code to ensure compatibility
+            // with Java big decimal logical type.
+            let file: File = File::open("./tests/bigdec.avro")?;
+            let mut reader = Reader::new(BufReader::new(&file))?;
+            let next_element = reader.next().await;
+            assert!(next_element.is_some());
+            let value = next_element.unwrap()?;
+            let bg = match value {
+                Value::Record(ref fields) => Ok(&fields[0].1),
+                other => Err(format!("Expected a Value::Record, got: {other:?}")),
+            }?;
+            let value_big_decimal = match bg {
+                Value::BigDecimal(val) => Ok(val),
+                other => Err(format!("Expected a Value::BigDecimal, got: {other:?}")),
+            }?;
+
+            let ref_value = BigDecimal::from_str("2.24")?;
+            assert_eq!(&ref_value, value_big_decimal);
+
+            Ok(())
+        }
     }
 }