[ISSUE #3154]🚀Implement NotificationProcessor (#3157)

Author: mxsm

rocketmq-broker/src/broker_runtime.rs

@@ -577,7 +577,7 @@ impl BrokerRuntime {
         ));
         self.inner.ack_message_processor = Some(ack_message_processor.clone());
 
-        let notification_processor = ArcMut::new(NotificationProcessor::default());
+        let notification_processor = NotificationProcessor::new(self.inner.clone());
         self.inner.notification_processor = Some(notification_processor.clone());
         BrokerRequestProcessor {
             send_message_processor: ArcMut::new(send_message_processor),
@@ -1335,7 +1335,7 @@ pub(crate) struct BrokerRuntimeInner<MS> {
     //Processor
     pop_message_processor: Option<ArcMut<PopMessageProcessor<MS>>>,
     ack_message_processor: Option<ArcMut<AckMessageProcessor<MS>>>,
-    notification_processor: Option<ArcMut<NotificationProcessor>>,
+    notification_processor: Option<ArcMut<NotificationProcessor<MS>>>,
     broker_attached_plugins: Vec<Arc<dyn BrokerAttachedPlugin>>,
 }
 
@@ -2147,7 +2147,7 @@ impl<MS: MessageStore> BrokerRuntimeInner<MS> {
         unsafe { self.ack_message_processor.as_ref().unwrap_unchecked() }
     }
 
-    pub fn notification_processor_unchecked(&self) -> &ArcMut<NotificationProcessor> {
+    pub fn notification_processor_unchecked(&self) -> &ArcMut<NotificationProcessor<MS>> {
         unsafe { self.notification_processor.as_ref().unwrap_unchecked() }
     }
 

rocketmq-broker/src/long_polling/notify_message_arriving_listener.rs

@@ -81,7 +81,7 @@ where
 
         self.broker_runtime_inner
             .notification_processor_unchecked()
-            .notify_message_arriving_full(
+            .notify_message_arriving(
                 topic.clone(),
                 queue_id,
                 tags_code,

rocketmq-broker/src/processor.rs

@@ -67,7 +67,7 @@ pub struct BrokerRequestProcessor<MS, TS> {
     pub(crate) pop_message_processor: ArcMut<PopMessageProcessor<MS>>,
     pub(crate) ack_message_processor: ArcMut<AckMessageProcessor<MS>>,
     pub(crate) change_invisible_time_processor: ArcMut<ChangeInvisibleTimeProcessor<MS>>,
-    pub(crate) notification_processor: ArcMut<NotificationProcessor>,
+    pub(crate) notification_processor: ArcMut<NotificationProcessor<MS>>,
     pub(crate) polling_info_processor: ArcMut<PollingInfoProcessor>,
     pub(crate) reply_message_processor: ArcMut<ReplyMessageProcessor<MS, TS>>,
     pub(crate) query_message_processor: ArcMut<QueryMessageProcessor<MS>>,