[ISSUE #1828]🚀Implement MQClientAPIImpl#ackMessageAsync

rocketmq-client/src/client_error.rs

@@ -41,6 +41,9 @@ pub enum MQClientError {
 
     #[error("{0}")]
     IllegalArgumentError(String),
+
+    #[error("{0}")]
+    CommonError(#[from] rocketmq_common::error::Error),
 }
 
 #[derive(Error, Debug)]

rocketmq-client/src/consumer/consumer_impl/default_mq_push_consumer_impl.rs

@@ -16,6 +16,7 @@
  */
 
 use std::collections::HashSet;
+use std::error::Error;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
@@ -63,6 +64,7 @@
 use crate::client_error::ClientErr;
 use crate::client_error::MQClientError;
 use crate::consumer::ack_callback::AckCallback;
+use crate::consumer::ack_result::AckResult;
 use crate::consumer::consumer_impl::consume_message_concurrently_service::ConsumeMessageConcurrentlyService;
 use crate::consumer::consumer_impl::consume_message_orderly_service::ConsumeMessageOrderlyService;
 use crate::consumer::consumer_impl::consume_message_pop_concurrently_service::ConsumeMessagePopConcurrentlyService;
@@ -1431,7 +1433,7 @@
             error!("The broker[{}] not exist", des_broker_name);
             return;
         }
-
+        let find_broker_result = find_broker_result.unwrap();
         let request_header = AckMessageRequestHeader {
             consumer_group: consumer_group.clone(),
             topic: CheetahString::from_string(
@@ -1449,8 +1451,29 @@
                 lo: None,
             }),
         };
-        //client_instance.mq_client_api_impl.as_mut().unwrap()
-        unimplemented!("ackAsync");
+        struct DefaultAckCallback;
+        impl AckCallback for DefaultAckCallback {
+            fn on_success(&self, _ack_result: AckResult) {}
+
+            fn on_exception(&self, _e: Box<dyn Error>) {}
+        }
+        match client_instance
+            .mq_client_api_impl
+            .as_mut()
+            .unwrap()
+            .ack_message_async(
+                &find_broker_result.broker_addr,
+                request_header,
+                ASYNC_TIMEOUT,
+                DefaultAckCallback,
+            )
+            .await
+        {
+            Ok(_) => {}
+            Err(e) => {
+                error!("ackAsync error: {}", e);
+            }
+        }
     }
 
     pub(crate) async fn change_pop_invisible_time_async(

rocketmq-client/src/implementation/mq_client_api_impl.rs

@@ -42,6 +42,7 @@
 use rocketmq_remoting::clients::RemotingClient;
 use rocketmq_remoting::code::request_code::RequestCode;
 use rocketmq_remoting::code::response_code::ResponseCode;
+use rocketmq_remoting::protocol::body::batch_ack_message_request_body::BatchAckMessageRequestBody;
 use rocketmq_remoting::protocol::body::check_client_request_body::CheckClientRequestBody;
 use rocketmq_remoting::protocol::body::get_consumer_listby_group_response_body::GetConsumerListByGroupResponseBody;
 use rocketmq_remoting::protocol::body::query_assignment_request_body::QueryAssignmentRequestBody;
@@ -50,6 +51,7 @@
 use rocketmq_remoting::protocol::body::response::lock_batch_response_body::LockBatchResponseBody;
 use rocketmq_remoting::protocol::body::set_message_request_mode_request_body::SetMessageRequestModeRequestBody;
 use rocketmq_remoting::protocol::body::unlock_batch_request_body::UnlockBatchRequestBody;
+use rocketmq_remoting::protocol::header::ack_message_request_header::AckMessageRequestHeader;
 use rocketmq_remoting::protocol::header::change_invisible_time_request_header::ChangeInvisibleTimeRequestHeader;
 use rocketmq_remoting::protocol::header::change_invisible_time_response_header::ChangeInvisibleTimeResponseHeader;
 use rocketmq_remoting::protocol::header::client_request_header::GetRouteInfoRequestHeader;
@@ -1688,6 +1690,67 @@
         }
         Ok(pop_result)
     }
+
+    pub async fn ack_message_async(
+        &self,
+        addr: &CheetahString,
+        request_header: AckMessageRequestHeader,
+        timeout_millis: u64,
+        ack_callback: impl AckCallback,
+    ) -> Result<()> {
+        self.ack_message_async_inner(
+            addr,
+            Some(request_header),
+            None,
+            timeout_millis,
+            ack_callback,
+        )
+        .await
+    }
+
+    pub(self) async fn ack_message_async_inner(
+        &self,
+        addr: &CheetahString,
+        request_header: Option<AckMessageRequestHeader>,
+        request_body: Option<BatchAckMessageRequestBody>,
+        timeout_millis: u64,
+        ack_callback: impl AckCallback,
+    ) -> Result<()> {
+        let request = if let Some(header) = request_header {
+            RemotingCommand::create_request_command(RequestCode::AckMessage, header)
+        } else {
+            let body = request_body.unwrap();
+            RemotingCommand::new_request(
+                RequestCode::BatchAckMessage,
+                body.encode().map_err(MQClientError::CommonError)?,
+            )
+        };
+        match self
+            .remoting_client
+            .invoke_async(Some(addr), request, timeout_millis)
+            .await
+        {
+            Ok(response) => {
+                let response_code = ResponseCode::from(response.code());
+                let ack_result = if response_code == ResponseCode::Success {
+                    AckResult {
+                        status: AckStatus::Ok,
+                        ..Default::default()
+                    }
+                } else {
+                    AckResult {
+                        status: AckStatus::NotExist,
+                        ..Default::default()
+                    }
+                };
+                ack_callback.on_success(ack_result);
+            }
+            Err(e) => {
+                ack_callback.on_exception(Box::new(e));
+            }
+        }
+        Ok(())
+    }
 }
 
 fn build_queue_offset_sorted_map(