rocketmq-client/src/consumer/consumer_impl/pop_process_queue.rs (4)

45-52: Refactor to eliminate redundant new method

The new method duplicates the functionality provided by the Default implementation. It's more idiomatic in Rust to use PopProcessQueue::default() instead of defining a separate new method with the same logic. Consider removing the new method to avoid redundancy.

Apply this diff to remove the new method:

 impl PopProcessQueue {
-    pub(crate) fn new() -> Self {
-        PopProcessQueue {
-            last_pop_timestamp: get_current_millis(),
-            wait_ack_counter: Arc::new(AtomicUsize::new(0)),
-            dropped: Arc::new(AtomicBool::new(false)),
-        }
-    }
     // Existing methods...
 }

Update instances where PopProcessQueue::new() is called to use PopProcessQueue::default() instead.

---

99-108: Implement the Debug trait for better debugging support

While the Display trait is implemented, leveraging the Debug trait is standard practice for structs, especially when using debugging tools or printing debug information.

Implement the Debug trait as follows:

impl std::fmt::Debug for PopProcessQueue {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PopProcessQueue")
            .field("last_pop_timestamp", &self.last_pop_timestamp)
            .field("wait_ack_counter", &self.wait_ack_counter.load(Ordering::Relaxed))
            .field("dropped", &self.dropped.load(Ordering::Relaxed))
            .finish()
    }
}

---

18-21: Simplify import statements for atomic types

Consolidate the import statements for atomic types for clarity and conciseness.

Replace:

use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;

With:

use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

---

35-43: Delegate Default implementation to existing new method

If the new method is retained, you can avoid code duplication by having the Default implementation call new.

Modify the Default implementation:

 impl Default for PopProcessQueue {
     fn default() -> Self {
-        PopProcessQueue {
-            last_pop_timestamp: get_current_millis(),
-            wait_ack_counter: Arc::new(AtomicUsize::new(0)),
-            dropped: Arc::new(AtomicBool::new(false)),
-        }
+        Self::new()
     }
 }

Configuration used: .coderabbit.yaml
Review profile: CHILL

rocketmq-client/src/consumer/consumer_impl/pop_process_queue.rs (2)

35-52: Consider reducing code duplication between Default and new

The new() method duplicates the exact same initialization as the Default implementation. Consider implementing one in terms of the other.

 impl PopProcessQueue {
     pub(crate) fn new() -> Self {
-        PopProcessQueue {
-            last_pop_timestamp: get_current_millis(),
-            wait_ack_counter: Arc::new(AtomicUsize::new(0)),
-            dropped: Arc::new(AtomicBool::new(false)),
-        }
+        Self::default()
     }
 }

---

111-168: Consider adding more edge cases and concurrent tests

While the current test coverage is good, consider adding:

1. Concurrent access tests to verify thread safety
2. Edge cases for message counter overflow
3. Tests for negative count handling in dec_found_msg

Example concurrent test:

#[test]
fn test_concurrent_operations() {
    use std::thread;
    let queue = Arc::new(PopProcessQueue::new());
    let queue_clone = queue.clone();
    
    let t1 = thread::spawn(move || {
        for _ in 0..1000 {
            queue_clone.inc_found_msg(1);
            queue_clone.dec_found_msg(1);
        }
    });
    
    let t2 = thread::spawn(move || {
        for _ in 0..1000 {
            queue.ack();
        }
    });
    
    t1.join().unwrap();
    t2.join().unwrap();
}

Configuration used: .coderabbit.yaml
Review profile: CHILL

 pub(crate) fn dec_found_msg(&self, count: isize) {
-    self.wait_ack_counter
-        .fetch_add(-count as usize, Ordering::Relaxed);
+    self.wait_ack_counter
+        .fetch_sub(count as usize, Ordering::Relaxed);
 }