[ISSUE #366]🚀Implement ReputMessageService

rocketmq-store/src/base/select_result.rs

@@ -15,16 +15,21 @@
  * limitations under the License.
  */
 
+use std::sync::Arc;
+
+use crate::log_file::mapped_file::default_impl::DefaultMappedFile;
+
 /// Represents the result of selecting a mapped buffer.
 pub struct SelectMappedBufferResult {
     /// The start offset.
-    pub start_offset: i64,
+    pub start_offset: u64,
     /// The ByteBuffer.
-    pub byte_buffer: Vec<u8>, // Using Vec<u8> as a simplified representation of ByteBuffer in Rust
+    //  pub byte_buffer: Vec<u8>, // Using Vec<u8> as a simplified representation of ByteBuffer in
+    // Rust
     /// The size.
     pub size: i32,
     /// The mapped file.
-    //pub mapped_file: Option<&'a dyn MappedFile>,
-    /// Indicates whether the buffer is in the cache.
+    pub mapped_file: Option<Arc<DefaultMappedFile>>,
+    /// Whether the buffer is in cache.
     pub is_in_cache: bool,
 }

rocketmq-store/src/consume_queue/mapped_file_queue.rs

@@ -143,6 +143,13 @@
         self.mapped_files.last().cloned()
     }
 
+    pub fn get_first_mapped_file(&self) -> Option<Arc<DefaultMappedFile>> {
+        if self.mapped_files.is_empty() {
+            return None;
+        }
+        self.mapped_files.first().cloned()
+    }
+
     pub fn get_last_mapped_file_mut_start_offset(
         &mut self,
         start_offset: u64,
@@ -273,6 +280,50 @@
             let _ = fs::remove_dir_all(path);
         }
     }
+
+    pub fn find_mapped_file_by_offset(
+        &self,
+        offset: i64,
+        return_first_on_not_found: bool,
+    ) -> Option<Arc<DefaultMappedFile>> {
+        let first_mapped_file = self.get_first_mapped_file();
+        let last_mapped_file = self.get_last_mapped_file();
+        if first_mapped_file.is_some() && last_mapped_file.is_some() {
+            if offset < first_mapped_file.as_ref().unwrap().get_file_from_offset() as i64
+                || offset
+                    >= last_mapped_file.as_ref().unwrap().get_file_from_offset() as i64
+                        + self.mapped_file_size as i64
+            {
+                None
+            } else {
+                let index = offset as usize / self.mapped_file_size as usize
+                    - first_mapped_file.as_ref().unwrap().get_file_from_offset() as usize
+                        / self.mapped_file_size as usize;
+                let target_file = self.mapped_files.get(index).cloned();
+                if target_file.is_some()
+                    && offset >= target_file.as_ref().unwrap().get_file_from_offset() as i64
+                {
+                    return target_file;
+                }
+                for index in 0..self.mapped_files.len() {
+                    let mapped_file = self.mapped_files.get(index).unwrap();
+                    if offset >= mapped_file.get_file_from_offset() as i64
+                        && offset
+                            < mapped_file.get_file_from_offset() as i64
+                                + self.mapped_file_size as i64
+                    {
+                        return Some(mapped_file.clone());
+                    }
+                }
+                if return_first_on_not_found {
+                    return first_mapped_file;
+                }
+                None
+            }
+        } else {
+            None
+        }
+    }
 }
 
 #[cfg(test)]

rocketmq-store/src/log_file.rs

@@ -54,4 +54,6 @@ pub trait RocketMQMessageStore: Clone {
     fn get_state_machine_version(&self) -> i64;
 
     async fn put_message(&mut self, msg: MessageExtBrokerInner) -> PutMessageResult;
+
+    fn truncate_files(&mut self, offset_to_truncate: i64) -> bool;
 }

rocketmq-store/src/log_file/commit_log.rs

@@ -48,6 +48,7 @@
         message_result::PutMessageResult,
         message_status_enum::{AppendMessageStatus, PutMessageStatus},
         put_message_context::PutMessageContext,
+        select_result::SelectMappedBufferResult,
         store_checkpoint::StoreCheckpoint,
         swappable::Swappable,
     },
@@ -360,7 +361,7 @@
         } else {
             warn!(
                 "The commitlog files are deleted, and delete the consume queue
-                          files"
+                           files"
             );
             self.mapped_file_queue.set_flushed_where(0);
             self.mapped_file_queue.set_committed_where(0);
@@ -539,7 +540,7 @@
         } else {
             warn!(
                 "The commitlog files are deleted, and delete the consume queue
-                          files"
+                           files"
             );
             self.mapped_file_queue.set_flushed_where(0);
             self.mapped_file_queue.set_committed_where(0);
@@ -551,6 +552,44 @@
     pub fn get_max_offset(&self) -> i64 {
         self.mapped_file_queue.get_max_offset()
     }
+
+    pub fn get_min_offset(&self) -> i64 {
+        match self.mapped_file_queue.get_first_mapped_file() {
+            None => -1,
+            Some(mapped_file) => {
+                if mapped_file.is_available() {
+                    mapped_file.get_file_from_offset() as i64
+                } else {
+                    self.roll_next_file(mapped_file.get_file_from_offset() as i64)
+                }
+            }
+        }
+    }
+
+    pub fn roll_next_file(&self, offset: i64) -> i64 {
+        let mapped_file_size = self.message_store_config.mapped_file_size_commit_log as i64;
+        offset + mapped_file_size - (offset % mapped_file_size)
+    }
+
+    pub fn get_data(&self, offset: i64) -> Option<SelectMappedBufferResult> {
+        self.get_data_with_option(offset, offset == 0)
+    }
+    pub fn get_data_with_option(
+        &self,
+        offset: i64,
+        return_first_on_not_found: bool,
+    ) -> Option<SelectMappedBufferResult> {
+        let mapped_file_size = self.message_store_config.mapped_file_size_commit_log as i64;
+        let mapped_file = self
+            .mapped_file_queue
+            .find_mapped_file_by_offset(offset, return_first_on_not_found);
+        if let Some(mapped_file) = mapped_file {
+            let pos = (offset % mapped_file_size) as i32;
+            DefaultMappedFile::select_mapped_buffer(mapped_file, pos)
+        } else {
+            None
+        }
+    }
 }
 
 fn generate_key(msg: &MessageExtBrokerInner) -> String {
@@ -561,7 +600,7 @@
     topic_queue_key
 }
 
-fn check_message_and_return_size(
+pub fn check_message_and_return_size(
     bytes: &mut Bytes,
     check_crc: bool,
     check_dup_info: bool,

rocketmq-store/src/log_file/mapped_file.rs

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-use std::{fs::File, io};
+use std::{fs::File, io, sync::Arc};
 
 use bytes::Bytes;
 use rocketmq_common::common::message::{
@@ -103,7 +103,7 @@ pub trait MappedFile {
 
     /// Selects a slice of the mapped byte buffer's sub-region behind the mapped file, starting at
     /// the given position.
-    fn select_mapped_buffer(&self, pos: usize) -> SelectMappedBufferResult;
+    fn select_mapped_buffer(self: Arc<Self>, pos: i32) -> Option<SelectMappedBufferResult>;
 
     /// Returns the mapped byte buffer behind the mapped file.
     fn get_mapped_byte_buffer(&self) -> bytes::Bytes;

rocketmq-store/src/log_file/mapped_file/default_impl.rs

@@ -20,10 +20,13 @@
     io::Write,
     path::PathBuf,
     ptr,
-    sync::atomic::{AtomicBool, AtomicI32, AtomicI64, Ordering},
+    sync::{
+        atomic::{AtomicBool, AtomicI32, AtomicI64, Ordering},
+        Arc,
+    },
 };
 
-use bytes::{BufMut, Bytes, BytesMut};
+use bytes::{Bytes, BytesMut};
 use memmap2::MmapMut;
 use rocketmq_common::common::message::{
     message_batch::MessageExtBatch, message_single::MessageExtBrokerInner,
@@ -212,7 +215,7 @@
     }
 
     fn is_available(&self) -> bool {
-        todo!()
+        self.reference_resource.available.load(Ordering::Relaxed)
     }
 
     fn append_message<AMC: AppendMessageCallback>(
@@ -324,8 +327,18 @@
         todo!()
     }
 
-    fn select_mapped_buffer(&self, pos: usize) -> SelectMappedBufferResult {
-        todo!()
+    fn select_mapped_buffer(self: Arc<Self>, pos: i32) -> Option<SelectMappedBufferResult> {
+        let read_position = self.get_read_position();
+        if pos < read_position && read_position > 0 && self.hold() {
+            Some(SelectMappedBufferResult {
+                start_offset: self.get_file_from_offset() + pos as u64,
+                size: read_position - pos,
+                mapped_file: Some(self),
+                is_in_cache: false,
+            })
+        } else {
+            None
+        }
     }
 
     fn get_mapped_byte_buffer(&self) -> Bytes {
@@ -349,8 +362,7 @@
         let read_end_position = pos + size;
         if read_end_position <= read_position as usize {
             if self.hold() {
-                let mut buffer = BytesMut::with_capacity(size);
-                buffer.put(&self.mmapped_file.lock()[pos..read_end_position]);
+                let buffer = BytesMut::from(&self.mmapped_file.lock()[pos..read_end_position]);
                 Some(buffer.freeze())
             } else {
                 debug!(

rocketmq-store/src/log_file/mapped_file/default_impl_refactor.rs

@@ -19,7 +19,10 @@
     fs::{File, OpenOptions},
     io::Write,
     path::PathBuf,
-    sync::atomic::{AtomicI32, Ordering},
+    sync::{
+        atomic::{AtomicI32, Ordering},
+        Arc,
+    },
 };
 
 use bytes::{Bytes, BytesMut};
@@ -281,7 +284,7 @@
         todo!()
     }
 
-    fn select_mapped_buffer(&self, pos: usize) -> SelectMappedBufferResult {
+    fn select_mapped_buffer(self: Arc<Self>, pos: i32) -> Option<SelectMappedBufferResult> {
         todo!()
     }