[improve][broker]tidy update subscriptions dispatcher rate-limiter
@@ -384,14 +384,6 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
this.dispatchRateLimiter = Optional.of(new DispatchRateLimiter(this, Type.TOPIC));
}
- // dispatch rate limiter for each subscription
- subscriptions.forEach((name, subscription) -> {
- Dispatcher dispatcher = subscription.getDispatcher();
- if (dispatcher != null) {
- dispatcher.initializeDispatchRateLimiterIfNeeded();
- }
- });
-
// dispatch rate limiter for each replicator
replicators.forEach((name, replicator) ->
replicator.initializeDispatchRateLimiterIfNeeded());
@@ -2409,19 +2401,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
producer.checkPermissionsAsync().thenRun(producer::checkEncryption)));
return FutureUtil.waitForAll(producerCheckFutures).thenCompose((__) -> {
- List<CompletableFuture<Void>> subscriptionCheckFutures = new ArrayList<>((int) subscriptions.size());
- subscriptions.forEach((subName, sub) -> {
- List<CompletableFuture<Void>> consumerCheckFutures = new ArrayList<>(sub.getConsumers().size());
- sub.getConsumers().forEach(consumer -> consumerCheckFutures.add(consumer.checkPermissionsAsync()));
- subscriptionCheckFutures.add(FutureUtil.waitForAll(consumerCheckFutures).thenRun(() -> {
- Dispatcher dispatcher = sub.getDispatcher();
- if (dispatcher != null) {
- dispatcher.getRateLimiter().ifPresent(DispatchRateLimiter::updateDispatchRate);
- }
- }));
- });
-
- return FutureUtil.waitForAll(subscriptionCheckFutures).thenCompose((___) -> {
+ return updateSubscriptionsDispatcherRateLimiter().thenCompose((___) -> {
replicators.forEach((name, replicator) ->
replicator.getRateLimiter().ifPresent(DispatchRateLimiter::updateDispatchRate)
);
@@ -3050,17 +3030,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
dispatchRateLimiter.ifPresent(DispatchRateLimiter::updateDispatchRate);
- List<CompletableFuture<Void>> consumerCheckFutures = new ArrayList<>();
- subscriptions.forEach((subName, sub) -> sub.getConsumers().forEach(consumer -> {
- consumerCheckFutures.add(consumer.checkPermissionsAsync().thenRun(() -> {
- Dispatcher dispatcher = sub.getDispatcher();
- if (dispatcher != null) {
- dispatcher.updateRateLimiter();
- }
- }));
- }));
-
- FutureUtil.waitForAll(consumerCheckFutures).thenRun(() -> {
+ updateSubscriptionsDispatcherRateLimiter().thenRun(() -> {
updatePublishDispatcher();
updateSubscribeRateLimiter();
replicators.forEach((name, replicator) -> replicator.getRateLimiter()
@@ -3083,6 +3053,21 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
});
}
+ private CompletableFuture<Void> updateSubscriptionsDispatcherRateLimiter() {
+ List<CompletableFuture<Void>> subscriptionCheckFutures = new ArrayList<>((int) subscriptions.size());
+ subscriptions.forEach((subName, sub) -> {
+ List<CompletableFuture<Void>> consumerCheckFutures = new ArrayList<>(sub.getConsumers().size());
+ sub.getConsumers().forEach(consumer -> consumerCheckFutures.add(consumer.checkPermissionsAsync()));
+ subscriptionCheckFutures.add(FutureUtil.waitForAll(consumerCheckFutures).thenRun(() -> {
+ Dispatcher dispatcher = sub.getDispatcher();
+ if (dispatcher != null) {
+ dispatcher.updateRateLimiter();
+ }
+ }));
+ });
+ return FutureUtil.waitForAll(subscriptionCheckFutures);
+ }
+
private void initializeTopicDispatchRateLimiterIfNeeded(TopicPolicies policies) {
synchronized (dispatchRateLimiterLock) {
if (!dispatchRateLimiter.isPresent() && policies.getDispatchRate() != null) {
