[GitHub] [pulsar] eolivelli commented on a diff in pull request #19153: [improve] PIP-241: add TopicEventListener / topic events for the BrokerService

[GitBox <gi...@apache.org>]

eolivelli commented on code in PR #19153:
URL: https://github.com/apache/pulsar/pull/19153#discussion_r1072464772


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -402,6 +407,60 @@ public BrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws
         this.bundlesQuotas = new BundlesQuotas(pulsar.getLocalMetadataStore());
     }
 
+    public void addTopicEventListener(TopicEventsListener... listeners) {
+        Objects.requireNonNull(listeners);
+        Arrays.stream(listeners)
+                .filter(x -> x != null)
+                .forEach(topicEventListeners::add);
+        getTopics().keys().forEach(topic ->
+                notifyTopicListeners(listeners, topic, TopicEvent.LOAD, EventStage.SUCCESS, null));
+    }
+
+    public void removeTopicEventListener(TopicEventsListener... listeners) {
+        Objects.requireNonNull(listeners);
+        Arrays.stream(listeners)
+                .filter(x -> x != null)
+                .forEach(topicEventListeners::remove);
+    }
+
+    private void notifyTopicListeners(String topic, TopicEvent event, EventStage stage) {
+        notifyTopicListeners(topic, event, stage, null);
+    }
+
+    private void notifyTopicListeners(String topic, TopicEvent event, EventStage stage, Throwable t) {
+        for (TopicEventsListener listener : topicEventListeners) {
+            try {
+                listener.handleEvent(topic, event, stage, t);
+            } catch (Throwable ex) {
+                log.error("TopicEventsListener exception while handling {}_{} for topic {}", event, stage, topic, ex);

Review Comment:
   nit:
   in case of InterruptedException we should call Thread.currentThread().interrupt();



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org