[ISSUE #349]🚀Add trait for DefaultMappedFile

rocketmq-common/src/common/message/message_single.rs

@@ -320,7 +320,7 @@
     pub message_ext_inner: MessageExt,
     pub properties_string: String,
     pub tags_code: i64,
-    pub encoded_buff: bytes::BytesMut,
+    pub encoded_buff: Option<bytes::BytesMut>,
     pub encode_completed: bool,
     pub version: MessageVersion,
 }

rocketmq-store/src/base/append_message_callback.rs

@@ -16,7 +16,7 @@
  */
 use std::{collections::HashMap, sync::Arc};
 
-use bytes::BytesMut;
+use bytes::{Bytes, BytesMut};
 use rocketmq_common::{
     common::{
         attribute::cq_type::CQType,
@@ -37,7 +37,7 @@
     config::message_store_config::MessageStoreConfig,
     log_file::{
         commit_log::{CommitLog, CRC32_RESERVED_LEN},
-        mapped_file::default_impl_refactor::LocalMappedFile,
+        mapped_file::MappedFile,
     },
 };
 
@@ -56,15 +56,24 @@
     /// # Returns
     ///
     /// The number of bytes written
-    fn do_append(
+    fn do_append<MF: MappedFile>(
         &self,
         file_from_offset: i64,
-        mapped_file: &LocalMappedFile,
+        mapped_file: &MF,
         max_blank: i32,
         msg: &mut MessageExtBrokerInner,
         put_message_context: &PutMessageContext,
     ) -> AppendMessageResult;
 
+    /*    fn do_append_back(
+        &self,
+        file_from_offset: i64,
+        mapped_file: &DefaultMappedFile,
+        max_blank: i32,
+        msg: &mut MessageExtBrokerInner,
+        put_message_context: &PutMessageContext,
+    ) -> AppendMessageResult;*/
+
     /// After batched message serialization, write MappedByteBuffer
     ///
     /// # Arguments
@@ -140,15 +149,15 @@
 
 #[allow(unused_variables)]
 impl AppendMessageCallback for DefaultAppendMessageCallback {
-    fn do_append(
+    fn do_append<MF: MappedFile>(
         &self,
         file_from_offset: i64,
-        mapped_file: &LocalMappedFile,
+        mapped_file: &MF,
         max_blank: i32,
         msg_inner: &mut MessageExtBrokerInner,
         put_message_context: &PutMessageContext,
     ) -> AppendMessageResult {
-        let mut pre_encode_buffer = msg_inner.encoded_buff.clone(); // Assuming get_encoded_buff returns Option<ByteBuffer>
+        let mut pre_encode_buffer = msg_inner.encoded_buff.take().unwrap(); // Assuming get_encoded_buff returns Option<ByteBuffer>
         let is_multi_dispatch_msg = self.message_store_config.enable_multi_dispatch
             && CommitLog::is_multi_dispatch_msg(msg_inner);
         if is_multi_dispatch_msg {
@@ -158,7 +167,7 @@
         }
 
         let msg_len = i32::from_le_bytes(pre_encode_buffer[0..4].try_into().unwrap());
-        let wrote_offset = file_from_offset + mapped_file.wrote_position() as i64;
+        let wrote_offset = file_from_offset + mapped_file.get_wrote_position() as i64;
 
         let msg_id =
             message_utils::build_message_id(msg_inner.message_ext_inner.store_host, wrote_offset);
@@ -175,6 +184,13 @@
         }
 
         if (msg_len + END_FILE_MIN_BLANK_LENGTH) > max_blank {
+            /*self.msg_store_item_memory.borrow_mut().clear();
+            self.msg_store_item_memory.borrow_mut().put_i32(max_blank);
+            self.msg_store_item_memory
+                .borrow_mut()
+                .put_i32(BLANK_MAGIC_CODE);
+            let bytes = self.msg_store_item_memory.borrow_mut().split().freeze();
+            mapped_file.append_message_bytes(&bytes);*/
             return AppendMessageResult {
                 status: AppendMessageStatus::EndOfFile,
                 wrote_offset,
@@ -199,8 +215,9 @@
         pre_encode_buffer[pos..(pos + 8)]
             .copy_from_slice(&msg_inner.store_timestamp().to_le_bytes());
 
-        msg_inner.encoded_buff = pre_encode_buffer;
-
+        // msg_inner.encoded_buff = pre_encode_buffer;
+        let bytes = Bytes::from(pre_encode_buffer);
+        mapped_file.append_message_bytes(&bytes);
         AppendMessageResult {
             status: AppendMessageStatus::PutOk,
             wrote_offset,

rocketmq-store/src/base/message_result.rs

@@ -109,9 +109,9 @@ impl PutMessageResult {
 }
 
 /// Represents the result of getting a message.
-pub struct GetMessageResult<'a> {
+pub struct GetMessageResult {
     /// The list of mapped buffer results.
-    pub message_mapped_list: Vec<SelectMappedBufferResult<'a>>,
+    pub message_mapped_list: Vec<SelectMappedBufferResult>,
     /// The list of message buffers.
     pub message_buffer_list: Vec<Vec<u8>>, /* Using Vec<u8> as a simplified representation of
                                             * ByteBuffer in Rust */

rocketmq-store/src/base/select_result.rs

@@ -15,18 +15,16 @@
  * limitations under the License.
  */
 
-use crate::log_file::mapped_file::MappedFileBak;
-
 /// Represents the result of selecting a mapped buffer.
-pub struct SelectMappedBufferResult<'a> {
+pub struct SelectMappedBufferResult {
     /// The start offset.
     pub start_offset: i64,
     /// The ByteBuffer.
     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 MappedFileBak>,
+    //pub mapped_file: Option<&'a dyn MappedFile>,
     /// Indicates whether the buffer is in the cache.
     pub is_in_cache: bool,
 }

rocketmq-store/src/consume_queue/mapped_file_queue.rs

@@ -26,7 +26,7 @@
 use tracing::info;
 
 use crate::{
-    log_file::mapped_file::default_impl_refactor::LocalMappedFile,
+    log_file::mapped_file::{default_impl::DefaultMappedFile, MappedFile},
     services::allocate_mapped_file_service::AllocateMappedFileService,
 };
 
@@ -37,7 +37,8 @@
     pub(crate) mapped_file_size: u64,
     //pub(crate) mapped_files: Arc<Mutex<Vec<LocalMappedFile>>>,
     //pub(crate) mapped_files: Vec<Arc<Mutex<LocalMappedFile>>>,
-    pub(crate) mapped_files: Vec<Arc<LocalMappedFile>>,
+    //pub(crate) mapped_files: Vec<Arc<LocalMappedFile>>,
+    pub(crate) mapped_files: Vec<Arc<DefaultMappedFile>>,
     //  pub(crate) mapped_files: Vec<LocalMappedFile>,
     pub(crate) allocate_mapped_file_service: Option<AllocateMappedFileService>,
 
@@ -127,7 +128,7 @@
             }
 
             let mapped_file =
-                LocalMappedFile::new(file.to_string_lossy().to_string(), self.mapped_file_size);
+                DefaultMappedFile::new(file.to_string_lossy().to_string(), self.mapped_file_size);
             // Set wrote, flushed, committed positions for mapped_file
 
             self.mapped_files.push(Arc::new(mapped_file));
@@ -153,7 +154,7 @@
     //     self.mapped_files.last()
     // }
 
-    pub fn get_last_mapped_file(&self) -> Option<Arc<LocalMappedFile>> {
+    pub fn get_last_mapped_file(&self) -> Option<Arc<DefaultMappedFile>> {
         if self.mapped_files.is_empty() {
             return None;
         }
@@ -164,7 +165,7 @@
         &mut self,
         start_offset: u64,
         need_create: bool,
-    ) -> Option<Arc<LocalMappedFile>> {
+    ) -> Option<Arc<DefaultMappedFile>> {
         let mut create_offset = -1i64;
         let file_size = self.mapped_file_size as i64;
         let mapped_file_last = self.get_last_mapped_file();
@@ -184,7 +185,7 @@
         mapped_file_last
     }
 
-    pub fn try_create_mapped_file(&mut self, create_offset: u64) -> Option<Arc<LocalMappedFile>> {
+    pub fn try_create_mapped_file(&mut self, create_offset: u64) -> Option<Arc<DefaultMappedFile>> {
         let next_file_path =
             PathBuf::from(self.store_path.clone()).join(offset_to_file_name(create_offset));
         let next_next_file_path = PathBuf::from(self.store_path.clone())
@@ -196,9 +197,9 @@
         &mut self,
         next_file_path: PathBuf,
         _next_next_file_path: PathBuf,
-    ) -> Option<Arc<LocalMappedFile>> {
+    ) -> Option<Arc<DefaultMappedFile>> {
         let mut mapped_file = match self.allocate_mapped_file_service {
-            None => LocalMappedFile::new(
+            None => DefaultMappedFile::new(
                 next_file_path.to_string_lossy().to_string(),
                 self.mapped_file_size,
             ),
@@ -215,7 +216,7 @@
         Some(inner)
     }
 
-    pub fn get_mapped_files(&self) -> Vec<Arc<LocalMappedFile>> {
+    pub fn get_mapped_files(&self) -> Vec<Arc<DefaultMappedFile>> {
         self.mapped_files.to_vec()
     }
 
@@ -259,7 +260,7 @@
         unimplemented!()
     }
 
-    pub(crate) fn delete_expired_file(&self, files: Vec<Option<Arc<LocalMappedFile>>>) {}
+    pub(crate) fn delete_expired_file(&self, files: Vec<Option<Arc<DefaultMappedFile>>>) {}
 }
 
 #[cfg(test)]

rocketmq-store/src/log_file/commit_log.rs

@@ -49,7 +49,7 @@
     },
     config::message_store_config::MessageStoreConfig,
     consume_queue::mapped_file_queue::MappedFileQueue,
-    log_file::mapped_file::default_impl_refactor::LocalMappedFile,
+    log_file::mapped_file::MappedFile,
     message_encoder::message_ext_encoder::MessageExtEncoder,
     message_store::default_message_store::{CommitLogDispatcherDefault, DefaultMessageStore},
     queue::ConsumeQueueStoreTrait,
@@ -182,7 +182,7 @@
         if let Some(result) = put_message_result {
             return result;
         }
-        msg.encoded_buff = encoder.byte_buf();
+        msg.encoded_buff = Some(encoder.byte_buf());
 
         //let mut mapped_file_guard = self.mapped_file_queue.write().await;
         // let mapped_file = match mapped_file_guard.get_last_mapped_file() {
@@ -195,12 +195,12 @@
             Some(mapped_file) => mapped_file,
         };
         let topic_queue_key = generate_key(&msg);
-        let mut put_message_context = PutMessageContext::new(topic_queue_key);
+        let put_message_context = PutMessageContext::new(topic_queue_key);
 
         let result = mapped_file.append_message(
             msg,
             self.append_message_callback.as_ref(),
-            &mut put_message_context,
+            &put_message_context,
         );
 
         match result.status {
@@ -268,7 +268,7 @@
             let do_dispatch = false;
             let mut current_pos = 0usize;
             loop {
-                let (msg, size) = self.get_simple_message_bytes(current_pos, mapped_file);
+                let (msg, size) = self.get_simple_message_bytes(current_pos, mapped_file.as_ref());
                 if msg.is_none() {
                     break;
                 }
@@ -349,10 +349,10 @@
         });*/
     }
 
-    fn get_simple_message_bytes(
+    fn get_simple_message_bytes<MF: MappedFile>(
         &self,
         position: usize,
-        mapped_file: &LocalMappedFile,
+        mapped_file: &MF,
     ) -> (Option<Bytes>, usize) {
         let mut bytes = mapped_file.get_bytes(position, 4);
         match bytes {