[ISSUE #312]✨Support send message and send message v2(request code:10,310)-5

Author: mxsm

rocketmq-broker/src/broker_runtime.rs

@@ -165,7 +165,9 @@ impl BrokerRuntime {
     fn initialize_message_store(&mut self) -> bool {
         if self.message_store_config.store_type == StoreType::LocalFile {
             info!("Use local file as message store");
-            self.message_store = Some(Arc::new(LocalFileMessageStore::new(self.message_store_config.clone())));
+            self.message_store = Some(Arc::new(LocalFileMessageStore::new(
+                self.message_store_config.clone(),
+            )));
         } else if self.message_store_config.store_type == StoreType::RocksDB {
             info!("Use RocksDB as message store");
         } else {

rocketmq-broker/src/processor/send_message_processor.rs

@@ -215,7 +215,6 @@ impl<MS: MessageStore + Send> SendMessageProcessor<MS> {
             );
         }
 
-        let mut response_header = SendMessageResponseHeader::default();
         let mut topic_config = self
             .topic_config_manager
             .select_topic_config(request_header.topic().as_str())
@@ -320,7 +319,6 @@ impl<MS: MessageStore + Send> SendMessageProcessor<MS> {
             response,
             &request,
             topic.as_str(),
-            &mut response_header,
             queue_id,
         )
     }
@@ -331,14 +329,14 @@ impl<MS: MessageStore + Send> SendMessageProcessor<MS> {
         response: RemotingCommand,
         request: &RemotingCommand,
         topic: &str,
-        response_header: &mut SendMessageResponseHeader,
         //  send_message_context: &mut SendMessageContext,
         // ctx: ConnectionHandlerContext,
         queue_id_int: i32,
         //begin_time_millis: i64,
         // mapping_context: TopicQueueMappingContext,
         //message_type: TopicMessageType,
     ) -> Option<RemotingCommand> {
+        let mut response_header = SendMessageResponseHeader::default();
         let mut response = response;
         let mut send_ok = false;
         match put_message_result.put_message_status() {
@@ -421,6 +419,7 @@ impl<MS: MessageStore + Send> SendMessageProcessor<MS> {
                     .logics_offset,
             );
             //response_header.set_transaction_id();
+            response = response.set_command_custom_header(Some(Box::new(response_header)));
             Some(response)
         } else {
             unimplemented!()

rocketmq-store/src/base/message_result.rs

@@ -43,6 +43,7 @@ pub struct AppendMessageResult {
     pub msg_num: i32,
 }
 
+#[derive(Default)]
 pub struct PutMessageResult {
     put_message_status: PutMessageStatus,
     append_message_result: Option<AppendMessageResult>,
@@ -64,16 +65,6 @@ impl Default for AppendMessageResult {
     }
 }
 
-impl Default for PutMessageResult {
-    fn default() -> Self {
-        Self{
-            put_message_status: Default::default(),
-            append_message_result: None,
-            remote_put: false,
-        }
-    }
-}
-
 impl PutMessageResult {
     pub fn new(
         put_message_status: PutMessageStatus,

rocketmq-store/src/config/message_store_config.rs

@@ -400,4 +400,4 @@ impl MessageStoreConfig {
         }
         self.store_path_commit_log.clone().unwrap()
     }
-}
+}

rocketmq-store/src/consume_queue/mapped_file_queue.rs

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
- use std::{
+use std::{
     fs,
     path::{Path, PathBuf},
     sync::Arc,
@@ -139,9 +139,7 @@ impl MappedFileQueue {
         if self.mapped_files.is_empty() {
             return None;
         }
-        self.mapped_files
-            .last()
-            .map_or(None, |last| Some(last.clone()))
+        self.mapped_files.last().cloned()
     }
 
     pub fn get_last_mapped_file_mut_start_offset(
@@ -157,7 +155,9 @@ impl MappedFileQueue {
                 create_offset = start_offset as i64 - (start_offset as i64 % file_size);
             }
             Some(ref value) => {
-                create_offset = value.lock().get_file_from_offset() as i64 + file_size;
+                if value.lock().is_full() {
+                    create_offset = value.lock().get_file_from_offset() as i64 + file_size
+                }
             }
         }
         if create_offset != -1 && need_create {

rocketmq-store/src/log_file/commit_log.rs

@@ -19,7 +19,9 @@ use std::{cell::Cell, ops::Deref, sync::Arc};
 
 use rocketmq_common::{
     common::{
-        attribute::cq_type::CQType, message::{message_single::MessageExtBrokerInner, MessageConst, MessageVersion}, mix_all
+        attribute::cq_type::CQType,
+        message::{message_single::MessageExtBrokerInner, MessageConst, MessageVersion},
+        mix_all,
     },
     utils::time_utils,
     CRC32Utils::crc32,
@@ -150,8 +152,11 @@ impl CommitLog {
             append_message_callback.do_append(mapped_file.file_from_offset() as i64, 0, &mut msg);
         mapped_file.append_data(msg.encoded_buff.clone(), false);*/
 
-        let result =
-            mapped_file.lock().append_message(msg, append_message_callback, &mut put_message_context);
+        let result = mapped_file.lock().append_message(
+            msg,
+            append_message_callback,
+            &mut put_message_context,
+        );
 
         match result.status {
             AppendMessageStatus::PutOk => {
@@ -179,13 +184,14 @@ impl CommitLog {
                 .starts_with(mix_all::RETRY_GROUP_TOPIC_PREFIX)
     }
 
-    pub fn get_message_num(&self,_msg_inner: &MessageExtBrokerInner) -> i16 {
-        let mut message_num = 1i16;
+    pub fn get_message_num(&self, _msg_inner: &MessageExtBrokerInner) -> i16 {
+        // let mut message_num = 1i16;
 
-        message_num
+        // message_num
+        1
     }
 
-    fn get_cq_type(&self,_msg_inner:MessageExtBrokerInner) -> CQType{
+    fn get_cq_type(&self, _msg_inner: MessageExtBrokerInner) -> CQType {
         CQType::SimpleCQ
     }
 }

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

@@ -28,10 +28,10 @@ use rocketmq_common::common::message::message_single::MessageExtBrokerInner;
 use tracing::error;
 
 use crate::base::{
-    append_message_callback::AppendMessageCallback, message_result::AppendMessageResult, message_status_enum::AppendMessageStatus, put_message_context::PutMessageContext
+    append_message_callback::AppendMessageCallback, message_result::AppendMessageResult,
+    message_status_enum::AppendMessageStatus, put_message_context::PutMessageContext,
 };
 
-
 pub struct LocalMappedFile {
     //file information
     file_name: String,
@@ -55,7 +55,8 @@ impl LocalMappedFile {
         let file = OpenOptions::new()
             .read(true)
             .write(true)
-            //.create(true)
+            .create(true)
+            .truncate(false)
             .open(&path_buf)
             .unwrap();
         file.set_len(file_size).unwrap();