[ISSUE #699]🚀Support pull message result handle (#701)

Author: TeslaRustor

rocketmq-broker/src/broker_runtime.rs

@@ -54,6 +54,7 @@ use crate::client::manager::producer_manager::ProducerManager;
 use crate::filter::manager::consumer_filter_manager::ConsumerFilterManager;
 use crate::hook::batch_check_before_put_message::BatchCheckBeforePutMessageHook;
 use crate::hook::check_before_put_message::CheckBeforePutMessageHook;
+use crate::offset::manager::broadcast_offset_manager::BroadcastOffsetManager;
 use crate::offset::manager::consumer_offset_manager::ConsumerOffsetManager;
 use crate::offset::manager::consumer_order_info_manager::ConsumerOrderInfoManager;
 use crate::out_api::broker_outer_api::BrokerOuterAPI;
@@ -372,6 +373,7 @@ impl BrokerRuntime {
             self.consumer_manager.clone(),
             self.consumer_filter_manager.clone(),
             Arc::new(self.consumer_offset_manager.clone()),
+            Arc::new(BroadcastOffsetManager::default()),
             self.message_store.as_ref().unwrap().clone(),
         );
 

rocketmq-broker/src/offset/manager/broadcast_offset_manager.rs

@@ -16,3 +16,18 @@
  */
 #[derive(Debug, Default)]
 pub struct BroadcastOffsetManager {}
+
+#[allow(unused_variables)]
+impl BroadcastOffsetManager {
+    pub fn query_init_offset(
+        &self,
+        topic: &str,
+        group_id: &str,
+        queue_id: i32,
+        client_id: &str,
+        request_offset: i64,
+        from_proxy: bool,
+    ) -> i64 {
+        unimplemented!()
+    }
+}

rocketmq-broker/src/processor/default_pull_message_result_handler.rs

@@ -90,14 +90,23 @@ impl PullMessageResultHandler for DefaultPullMessageResultHandler {
             &mut response,
             client_address.as_str(),
         );
-        /*        self.execute_consume_message_hook_before(
+        self.execute_consume_message_hook_before(
             &request,
             &request_header,
             &get_message_result,
             broker_allow_suspend,
             From::from(response.code()),
-        );*/
-
+        );
+        /*let response_header = response.read_custom_header_mut::<PullMessageResponseHeader>();
+        let rewrite_result = PullMessageProcessor::rewrite_response_for_static_topic(
+            &request_header,
+            response_header.unwrap(),
+            &mut mapping_context,
+            ResponseCode::from(response.code()),
+        );
+        if rewrite_result.is_some() {
+            return rewrite_result;
+        }*/
         None
     }
 }