[ISSUE #1480]⚡️Optimize SendMessageProcessor error handle🔥

rocketmq-broker/src/broker_error.rs

@@ -35,6 +35,32 @@
     ClientError(#[from] rocketmq_client_rust::client_error::MQClientError),
 }
 
+impl From<BrokerError> for rocketmq_remoting::remoting_error::RemotingError {
+    fn from(value: BrokerError) -> Self {
+        match value {
+            BrokerError::BrokerRemotingError(e) => e,
+            BrokerError::BrokerCommonError(e) => {
+                rocketmq_remoting::remoting_error::RemotingError::RemoteError(format!("{}", e))
+            }
+            BrokerError::MQBrokerError(code, message, _) => {
+                rocketmq_remoting::remoting_error::RemotingError::RemoteError(format!(
+                    "CODE:{}, Message:{}",
+                    code, message
+                ))
+            }
+            BrokerError::IllegalArgumentError(e) => {
+                rocketmq_remoting::remoting_error::RemotingError::RemoteError(e)
+            }
+            BrokerError::ClientError(e) => {
+                rocketmq_remoting::remoting_error::RemotingError::RemotingCommandError(format!(
+                    "{}",
+                    e
+                ))
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use rocketmq_remoting::remoting_error::RemotingError;

rocketmq-broker/src/processor.rs

@@ -117,9 +117,10 @@
             | RequestCode::SendMessageV2
             | RequestCode::SendBatchMessage
             | RequestCode::ConsumerSendMsgBack => {
-                self.send_message_processor
+                return self
+                    .send_message_processor
                     .process_request(channel, ctx, request_code, request)
-                    .await
+                    .await;
             }
 
             RequestCode::SendReplyMessage | RequestCode::SendReplyMessageV2 => {

rocketmq-broker/src/processor/send_message_processor.rs

@@ -110,13 +110,13 @@
         ctx: ConnectionHandlerContext,
         request_code: RequestCode,
         request: RemotingCommand,
-    ) -> Option<RemotingCommand> {
+    ) -> rocketmq_remoting::Result<Option<RemotingCommand>> {
         match request_code {
             RequestCode::ConsumerSendMsgBack => {
                 //need to optimize
                 self.inner
                     .consumer_send_msg_back(&channel, &ctx, &request)
-                    .unwrap()
+                    .map_err(Into::into)
             }
             _ => {
                 let mut request_header = parse_request_header(&request, request_code).unwrap(); //need to optimize
@@ -129,7 +129,7 @@
                     &mapping_context,
                 );
                 if let Some(rewrite_result) = rewrite_result {
-                    return Some(rewrite_result);
+                    return Ok(Some(rewrite_result));
                 }
 
                 let send_message_context =
@@ -145,28 +145,30 @@
                     };
                 if request_header.batch.is_none() || !request_header.batch.unwrap() {
                     //handle single message
-                    self.send_message(
-                        &channel,
-                        &ctx,
-                        request,
-                        send_message_context,
-                        request_header,
-                        mapping_context,
-                        execute_send_message_hook_after,
-                    )
-                    .await
+                    Ok(self
+                        .send_message(
+                            &channel,
+                            &ctx,
+                            request,
+                            send_message_context,
+                            request_header,
+                            mapping_context,
+                            execute_send_message_hook_after,
+                        )
+                        .await)
                 } else {
                     //handle batch message
-                    self.send_batch_message(
-                        &channel,
-                        &ctx,
-                        request,
-                        send_message_context,
-                        request_header,
-                        mapping_context,
-                        execute_send_message_hook_after,
-                    )
-                    .await
+                    Ok(self
+                        .send_batch_message(
+                            &channel,
+                            &ctx,
+                            request,
+                            send_message_context,
+                            request_header,
+                            mapping_context,
+                            execute_send_message_hook_after,
+                        )
+                        .await)
                 }
             }
         }