[ISSUE #972]🔥Supports client clusting consume🚀

mxsm · Sep 20, 2024 · 4edbba7 · 4edbba7
1 parent 887e9f6
commit 4edbba7
Show file tree

Hide file tree

Showing 38 changed files with 2,674 additions and 186 deletions.
diff --git a/rocketmq-client/examples/quickstart/consumer.rs b/rocketmq-client/examples/quickstart/consumer.rs
@@ -27,7 +27,7 @@ pub const MESSAGE_COUNT: usize = 1;
 pub const CONSUMER_GROUP: &str = "please_rename_unique_group_name_4";
 pub const DEFAULT_NAMESRVADDR: &str = "127.0.0.1:9876";
 pub const TOPIC: &str = "TopicTest";
-pub const TAG: &str = "TagA";
+pub const TAG: &str = "*";
 
 #[rocketmq::main]
 pub async fn main() -> Result<()> {
@@ -53,8 +53,8 @@ pub struct MyMessageListener;
 impl MessageListenerConcurrently for MyMessageListener {
     fn consume_message(
         &self,
-        msgs: Vec<MessageExt>,
-        _context: ConsumeConcurrentlyContext,
+        msgs: &[&MessageExt],
+        _context: &ConsumeConcurrentlyContext,
     ) -> Result<ConsumeConcurrentlyStatus> {
         for msg in msgs {
             println!("Receive message: {:?}", msg);

diff --git a/rocketmq-client/src/base/client_config.rs b/rocketmq-client/src/base/client_config.rs
@@ -142,15 +142,18 @@ impl ClientConfig {
         )
     }
 
-    pub fn queue_with_namespace(&mut self, queue: &mut MessageQueue) {
+    pub fn queue_with_namespace(&mut self, queue: MessageQueue) -> MessageQueue {
         if let Some(namespace) = self.get_namespace() {
             if !namespace.is_empty() {
-                queue.set_topic(NamespaceUtil::wrap_namespace(
+                let mut message_queue = queue.clone();
+                message_queue.set_topic(NamespaceUtil::wrap_namespace(
                     namespace.as_str(),
                     queue.get_topic(),
                 ));
+                return message_queue;
             }
         }
+        queue
     }
 
     pub fn get_namespace(&mut self) -> Option<String> {

diff --git a/rocketmq-client/src/consumer.rs b/rocketmq-client/src/consumer.rs
@@ -24,8 +24,8 @@ pub mod message_selector;
 pub mod mq_consumer;
 pub(crate) mod mq_consumer_inner;
 pub mod mq_push_consumer;
-mod pull_callback;
-mod pull_result;
-mod pull_status;
+pub(crate) mod pull_callback;
+pub(crate) mod pull_result;
+pub(crate) mod pull_status;
 pub mod rebalance_strategy;
-mod store;
+pub(crate) mod store;
diff --git a/rocketmq-client/src/consumer/consumer_impl.rs b/rocketmq-client/src/consumer/consumer_impl.rs
@@ -29,6 +29,7 @@ pub(crate) mod process_queue;
 pub(crate) mod pull_api_wrapper;
 pub(crate) mod pull_message_service;
 pub(crate) mod pull_request;
+pub(crate) mod pull_request_ext;
 pub(crate) mod re_balance;
 
 pub(crate) static PULL_MAX_IDLE_TIME: Lazy<u64> = Lazy::new(|| {