conflict resolve

Author: Zakir032002

rust/fory-core/src/buffer.rs

@@ -1167,3 +1167,5 @@ impl<'a> Reader<'a> {
 
 #[allow(clippy::needless_lifetimes)]
 unsafe impl<'a> Send for Reader<'a> {}
+#[allow(clippy::needless_lifetimes)]
+unsafe impl<'a> Sync for Reader<'a> {}

rust/fory-core/src/fory.rs

@@ -982,8 +982,17 @@ impl Fory {
                 // SAFETY: same invariant as Reader::from_stream and fill_to:
                 //   bf points into Box-owned stream buffer, owned by reader.stream,
                 //   which lives as long as reader.
-                if let Some(ref s) = reader.stream {
+                if let Some(ref mut s) = reader.stream {
+                    // Sync stream's read_pos with the reader cursor position
+                    // before shrinking — the detached reader may have advanced
+                    // cursor without updating stream.read_pos.
+                    let _ = s.set_reader_index(reader.cursor);
+                    // Mirror C++ StreamShrinkGuard: compact consumed bytes after
+                    // deserialization to prevent unbounded buffer growth on
+                    // long-lived streams.
+                    s.shrink_buffer();
                     reader.bf = unsafe { std::slice::from_raw_parts(s.data(), s.size()) };
+                    reader.cursor = s.reader_index();
                 }
                 result
             } else {
@@ -1027,7 +1036,11 @@ impl Fory {
             let reader = Reader::from_stream(stream);
             context.attach_reader(reader);
             let result = self.deserialize_with_context(context);
-            context.detach_reader();
+            // Mirror C++ StreamShrinkGuard: shrink_buffer on detach.
+            let mut returned = context.detach_reader();
+            if let Some(ref mut s) = returned.stream {
+                s.shrink_buffer();
+            }
             result
         })
     }

rust/fory-core/src/stream.rs

@@ -34,6 +34,8 @@ pub struct ForyStreamBuf {
     valid_len: usize,
     /// Current read cursor — equivalent of `gptr() - eback()`
     read_pos: usize,
+    /// Initial capacity for shrink_buffer target — mirrors C++ `initial_buffer_size_`
+    initial_buffer_size: usize,
 }
 
 impl ForyStreamBuf {
@@ -51,6 +53,7 @@ impl ForyStreamBuf {
             buffer,
             valid_len: 0,
             read_pos: 0,
+            initial_buffer_size: cap,
         }
     }
 
@@ -198,6 +201,44 @@ impl ForyStreamBuf {
     pub fn is_stream_backed(&self) -> bool {
         true
     }
+
+    /// Compact consumed bytes and optionally shrink capacity.
+    ///
+    /// Mirrors C++ `ForyInputStream::shrink_buffer()` exactly:
+    /// 1. Memmove remaining bytes to front of buffer
+    /// 2. Reset read_pos = 0, valid_len = remaining
+    /// 3. If capacity > initial_buffer_size and utilization is low,
+    ///    shrink back toward initial size
+    pub fn shrink_buffer(&mut self) {
+        let remaining = self.remaining();
+
+        // Phase 1: compact — memmove remaining data to front
+        if self.read_pos > 0 {
+            if remaining > 0 {
+                self.buffer.copy_within(self.read_pos..self.valid_len, 0);
+            }
+            self.read_pos = 0;
+            self.valid_len = remaining;
+        }
+
+        // Phase 2: optionally shrink capacity back toward initial_buffer_size
+        let current_capacity = self.buffer.len();
+        let mut target_capacity = current_capacity;
+
+        if current_capacity > self.initial_buffer_size {
+            if remaining == 0 {
+                target_capacity = self.initial_buffer_size;
+            } else if remaining <= current_capacity / 4 {
+                let doubled = remaining.saturating_mul(2).max(1);
+                target_capacity = self.initial_buffer_size.max(doubled);
+            }
+        }
+
+        if target_capacity < current_capacity {
+            self.buffer.truncate(target_capacity);
+            self.buffer.shrink_to_fit();
+        }
+    }
 }
 
 #[cfg(test)]

rust/tests/tests/stream_test.rs

@@ -20,8 +20,13 @@ mod stream_tests {
     use fory_core::buffer::Reader;
     use fory_core::stream::ForyStreamBuf;
     use fory_core::Fory;
+    use std::fmt::Debug;
     use std::io::Cursor;
 
+    // ========================================================================
+    // OneByteStream — mirrors C++ OneByteStreamBuf / OneByteIStream
+    // Delivers exactly 1 byte per read() call for maximum streaming stress.
+    // ========================================================================
     struct OneByte(Cursor<Vec<u8>>);
     impl std::io::Read for OneByte {
         fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
@@ -39,22 +44,57 @@ mod stream_tests {
         }
     }
 
+    // ========================================================================
+    // Deserialize helper — per maintainer requirement:
+    //   "Create a Deserialize help methods in tests, then use that instead of
+    //    fory.Deserialize for deserialization, and in the Deserialize test
+    //    helper, first deserialize from bytes, then wrap it into a
+    //    OneByteStream to deserialize it to ensure deserialization works."
+    // ========================================================================
+    fn deserialize_helper<T>(fory: &Fory, bytes: &[u8]) -> T
+    where
+        T: fory_core::Serializer + fory_core::ForyDefault + PartialEq + Debug,
+    {
+        // Path 1: deserialize from bytes (standard in-memory path)
+        let from_bytes: T = fory.deserialize(bytes).expect("bytes deserialize failed");
+
+        // Path 2: deserialize from OneByteStream (streaming path)
+        let from_stream: T = fory
+            .deserialize_from_stream(OneByte(Cursor::new(bytes.to_vec())))
+            .expect("stream deserialize failed");
+
+        // Assert both paths produce the same result
+        assert_eq!(
+            from_bytes, from_stream,
+            "bytes vs stream deserialization mismatch"
+        );
+
+        from_bytes
+    }
+
+    // ========================================================================
+    // Test: PrimitiveAndStringRoundTrip
+    // Mirrors C++ StreamSerializationTest::PrimitiveAndStringRoundTrip
+    // ========================================================================
     #[test]
-    fn test_primitive_stream_roundtrip() {
+    fn test_primitive_and_string_round_trip() {
         let fory = Fory::default();
+
+        // i64 round-trip
         let bytes = fory.serialize(&-9876543212345i64).unwrap();
-        let result: i64 = fory
-            .deserialize_from_stream(OneByte(Cursor::new(bytes)))
-            .unwrap();
+        let result = deserialize_helper::<i64>(&fory, &bytes);
         assert_eq!(result, -9876543212345i64);
 
+        // String round-trip (with unicode)
         let bytes = fory.serialize(&"stream-hello-世界".to_string()).unwrap();
-        let result: String = fory
-            .deserialize_from_stream(OneByte(Cursor::new(bytes)))
-            .unwrap();
+        let result = deserialize_helper::<String>(&fory, &bytes);
         assert_eq!(result, "stream-hello-世界");
     }
 
+    // ========================================================================
+    // Test: SequentialDeserializeFromSingleStream
+    // Mirrors C++ StreamSerializationTest::SequentialDeserializeFromSingleStream
+    // ========================================================================
     #[test]
     fn test_sequential_stream_reads() {
         let fory = Fory::default();
@@ -74,6 +114,10 @@ mod stream_tests {
         assert_eq!(third, 99);
     }
 
+    // ========================================================================
+    // Test: TruncatedStreamReturnsError
+    // Mirrors C++ StreamSerializationTest::TruncatedStreamReturnsError
+    // ========================================================================
     #[test]
     fn test_truncated_stream_returns_error() {
         let fory = Fory::default();
@@ -82,4 +126,59 @@ mod stream_tests {
         let result: Result<String, _> = fory.deserialize_from_stream(Cursor::new(bytes));
         assert!(result.is_err());
     }
+
+    // ========================================================================
+    // Test: ShrinkBuffer compacts consumed bytes
+    // Validates the C++ shrink_buffer() behavior is correctly implemented
+    // ========================================================================
+    #[test]
+    fn test_shrink_buffer_compacts_consumed_bytes() {
+        let fory = Fory::default();
+
+        // Serialize multiple values into a single buffer
+        let mut bytes = Vec::new();
+        fory.serialize_to(&mut bytes, &42i32).unwrap();
+        fory.serialize_to(&mut bytes, &"shrink-test".to_string())
+            .unwrap();
+        fory.serialize_to(&mut bytes, &100i64).unwrap();
+
+        // Use a small initial buffer to force multiple fills
+        let mut reader =
+            Reader::from_stream(ForyStreamBuf::with_capacity(OneByte(Cursor::new(bytes)), 4));
+
+        // After each deserialize_from, shrink_buffer should compact the stream.
+        let first: i32 = fory.deserialize_from(&mut reader).unwrap();
+        assert_eq!(first, 42);
+
+        let second: String = fory.deserialize_from(&mut reader).unwrap();
+        assert_eq!(second, "shrink-test");
+
+        let third: i64 = fory.deserialize_from(&mut reader).unwrap();
+        assert_eq!(third, 100);
+    }
+
+    // ========================================================================
+    // Test: Additional primitive types through deserialize_helper
+    // ========================================================================
+    #[test]
+    fn test_additional_primitive_types() {
+        let fory = Fory::default();
+
+        // bool
+        let bytes = fory.serialize(&true).unwrap();
+        assert_eq!(deserialize_helper::<bool>(&fory, &bytes), true);
+
+        // i32
+        let bytes = fory.serialize(&-42i32).unwrap();
+        assert_eq!(deserialize_helper::<i32>(&fory, &bytes), -42i32);
+
+        // f64
+        let bytes = fory.serialize(&3.14159f64).unwrap();
+        assert_eq!(deserialize_helper::<f64>(&fory, &bytes), 3.14159f64);
+
+        // Vec<i32>
+        let vec = vec![1i32, 2, 3, 5, 8];
+        let bytes = fory.serialize(&vec).unwrap();
+        assert_eq!(deserialize_helper::<Vec<i32>>(&fory, &bytes), vec);
+    }
 }