[ISSUE #218]📌Implement DefaultMappedFile initialization (#219)

* [ISSUE #218]📌Implement DefaultMappedFile initialization

* fix ci error

* fix test case error

Author: mxsm

rocketmq-store/Cargo.toml

@@ -56,5 +56,7 @@ num_cpus.workspace = true
 tempfile = "3.10.0"
 log = "0.4.20"
 
+memmap2 = "0.9.4"
+
 [dev-dependencies]
 tempfile = "3.10.0"

rocketmq-store/src/base.rs

@@ -15,9 +15,13 @@
  * limitations under the License.
  */
 
+pub mod append_message_callback;
+pub mod compaction_append_msg_callback;
 pub(crate) mod dispatch_request;
 pub mod message_result;
 pub mod message_status_enum;
+pub mod put_message_context;
 pub mod select_result;
 pub mod store_enum;
 pub mod swappable;
+pub mod transient_store_pool;

rocketmq-store/src/base/append_message_callback.rs

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+use rocketmq_common::common::message::{
+    message_batch::MessageExtBatch, message_single::MessageExtBrokerInner,
+};
+
+use crate::base::{message_result::AppendMessageResult, put_message_context::PutMessageContext};
+
+/// Write messages callback interface
+pub trait AppendMessageCallback {
+    /// After message serialization, write MappedByteBuffer
+    ///
+    /// # Arguments
+    ///
+    /// * `file_from_offset` - The offset of the file
+    /// * `byte_buffer` - The buffer to write
+    /// * `max_blank` - The maximum blank space
+    /// * `msg` - The message to write
+    /// * `put_message_context` - The context of putting message
+    ///
+    /// # Returns
+    ///
+    /// The number of bytes written
+    fn do_append(
+        &self,
+        file_from_offset: i64,
+        byte_buffer: &mut [u8],
+        max_blank: i32,
+        msg: &MessageExtBrokerInner,
+        put_message_context: &PutMessageContext,
+    ) -> AppendMessageResult;
+
+    /// After batched message serialization, write MappedByteBuffer
+    ///
+    /// # Arguments
+    ///
+    /// * `file_from_offset` - The offset of the file
+    /// * `byte_buffer` - The buffer to write
+    /// * `max_blank` - The maximum blank space
+    /// * `message_ext_batch` - The batched message to write
+    /// * `put_message_context` - The context of putting message
+    ///
+    /// # Returns
+    ///
+    /// The number of bytes written
+    fn do_append_batch(
+        &self,
+        file_from_offset: i64,
+        byte_buffer: &mut [u8],
+        max_blank: i32,
+        message_ext_batch: &MessageExtBatch,
+        put_message_context: &PutMessageContext,
+    ) -> AppendMessageResult;
+}

rocketmq-store/src/base/compaction_append_msg_callback.rs

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+use crate::base::message_result::AppendMessageResult;
+
+/// Callback interface for compaction append message
+pub trait CompactionAppendMsgCallback {
+    /// Append messages during compaction
+    ///
+    /// # Arguments
+    ///
+    /// * `bb_dest` - The destination buffer to append to
+    /// * `file_from_offset` - The offset of the file
+    /// * `max_blank` - The maximum blank space
+    /// * `bb_src` - The source buffer containing the message to be appended
+    ///
+    /// # Returns
+    ///
+    /// The result of the append operation
+    fn do_append(
+        &self,
+        bb_dest: &mut bytes::Bytes,
+        file_from_offset: i64,
+        max_blank: i32,
+        bb_src: &mut bytes::Bytes,
+    ) -> AppendMessageResult;
+}

rocketmq-store/src/base/put_message_context.rs

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#[derive(Debug, Clone, Default)]
+pub struct PutMessageContext {
+    topic_queue_table_key: String,
+    phy_pos: Vec<i64>,
+    batch_size: i32,
+}
+
+impl PutMessageContext {
+    pub fn new(topic_queue_table_key: String) -> Self {
+        PutMessageContext {
+            topic_queue_table_key,
+            phy_pos: Vec::new(),
+            batch_size: 0,
+        }
+    }
+
+    pub fn get_topic_queue_table_key(&self) -> &str {
+        &self.topic_queue_table_key
+    }
+
+    pub fn get_phy_pos(&self) -> &[i64] {
+        &self.phy_pos
+    }
+
+    pub fn phy_pos(&mut self, phy_pos: Vec<i64>) {
+        self.phy_pos = phy_pos;
+    }
+
+    pub fn get_batch_size(&self) -> i32 {
+        self.batch_size
+    }
+
+    pub fn batch_size(&mut self, batch_size: i32) {
+        self.batch_size = batch_size;
+    }
+}

rocketmq-store/src/base/transient_store_pool.rs

@@ -0,0 +1,18 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+pub struct TransientStorePool;

rocketmq-store/src/config.rs

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-mod broker_role;
-mod flush_disk_type;
+pub mod broker_role;
+pub mod flush_disk_type;
 pub mod message_store_config;
 pub(crate) mod store_path_config_helper;

rocketmq-store/src/consume_queue/mapped_file_queue.rs

@@ -118,8 +118,8 @@ mod tests {
     #[test]
     fn test_load_with_files() {
         let temp_dir = tempfile::tempdir().unwrap();
-        let file1_path = temp_dir.path().join("file1.txt");
-        let file2_path = temp_dir.path().join("file2.txt");
+        let file1_path = temp_dir.path().join("1111");
+        let file2_path = temp_dir.path().join("2222");
         fs::File::create(&file1_path).unwrap();
         fs::File::create(&file2_path).unwrap();
 
@@ -138,7 +138,7 @@ mod tests {
     #[test]
     fn test_load_with_empty_file() {
         let temp_dir = tempfile::tempdir().unwrap();
-        let file_path = temp_dir.path().join("empty_file.txt");
+        let file_path = temp_dir.path().join("1111");
         fs::File::create(&file_path).unwrap();
 
         let mut queue = MappedFileQueue {
@@ -174,7 +174,7 @@ mod tests {
     #[test]
     fn test_load_with_correct_file() {
         let temp_dir = tempfile::tempdir().unwrap();
-        let file_path = temp_dir.path().join("correct_file.txt");
+        let file_path = temp_dir.path().join("1111");
         fs::write(&file_path, vec![0u8; 1024]).unwrap();
 
         let mut queue = MappedFileQueue {