Review: Delay AMQP consumer status retrieval.

Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
eclipse-ditto · Oct 18, 2021 · 4ad1ec0 · 4ad1ec0
1 parent b98226d
commit 4ad1ec0
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 12 deletions.
diff --git a/.../src/main/java/org/eclipse/ditto/connectivity/service/messaging/amqp/AmqpClientActor.java b/.../src/main/java/org/eclipse/ditto/connectivity/service/messaging/amqp/AmqpClientActor.java
@@ -252,8 +252,10 @@ public void postStop() {
 
     @Override
     protected Set<Pattern> getExcludedAddressReportingChildNamePatterns() {
-        final Set<Pattern> excludedChildNamePatterns = new HashSet<>(super.getExcludedAddressReportingChildNamePatterns());
-        excludedChildNamePatterns.add(Pattern.compile(Pattern.quote(JMSConnectionHandlingActor.ACTOR_NAME_PREFIX) + ".*"));
+        final Set<Pattern> excludedChildNamePatterns =
+                new HashSet<>(super.getExcludedAddressReportingChildNamePatterns());
+        excludedChildNamePatterns.add(
+                Pattern.compile(Pattern.quote(JMSConnectionHandlingActor.ACTOR_NAME_PREFIX) + ".*"));
         return excludedChildNamePatterns;
     }
 
@@ -427,18 +429,17 @@ private CompletionStage<Done> startCommandConsumers(final List<ConsumerData> con
         if (isConsuming()) {
             stopCommandConsumers();
             // wait a fraction of the configured timeout before asking to allow the consumer to stabilize
-            final CompletionStage<Done> identity =
-                    new CompletableFuture<Done>().completeOnTimeout(Done.getInstance(),
+            final CompletionStage<Object> identity =
+                    new CompletableFuture<>().completeOnTimeout(Done.getInstance(),
                             initialConsumerResourceStatusAskTimeout.toMillis() / 2, TimeUnit.MILLISECONDS);
-            final CompletionStage<Done> completionStage = consumers.stream()
+            final CompletionStage<Object> completionStage = consumers.stream()
                     .map(consumer -> startCommandConsumer(consumer, getInboundMappingSink(), jmsActor))
-                    .map(ref -> Patterns.ask(ref, RetrieveAddressStatus.getInstance(),
-                            initialConsumerResourceStatusAskTimeout).thenApply(result -> Done.getInstance()))
-                    .reduce(identity, (done, reply) -> done.thenCompose(result -> done))
-                    .exceptionally(t -> Done.getInstance());
+                    .map(ref -> identity.thenCompose(done -> Patterns.ask(ref, RetrieveAddressStatus.getInstance(),
+                            initialConsumerResourceStatusAskTimeout)))
+                    .reduce(identity, (done, reply) -> done.thenCombine(reply, (x, y) -> x));
             connectionLogger.success("Subscriptions {0} initialized successfully.", consumers);
             logger.info("Subscribed Connection <{}> to sources: {}", connectionId(), consumers);
-            return completionStage;
+            return completionStage.thenApply(object -> Done.getInstance()).exceptionally(t -> Done.getInstance());
         } else {
             logger.debug("Not starting consumers, no sources were configured.");
             return CompletableFuture.completedStage(Done.getInstance());

diff --git a/.../test/java/org/eclipse/ditto/connectivity/service/messaging/amqp/AmqpClientActorTest.java b/.../test/java/org/eclipse/ditto/connectivity/service/messaging/amqp/AmqpClientActorTest.java
@@ -142,8 +142,6 @@ public final class AmqpClientActorTest extends AbstractBaseClientActorTest {
     private static final JMSException JMS_EXCEPTION = new JMSException("FAIL");
     private static final URI DUMMY = URI.create("amqp://test:1234");
     private static final ConnectionId CONNECTION_ID = TestConstants.createRandomConnectionId();
-    private static final ConnectionFailedException SESSION_EXCEPTION =
-            ConnectionFailedException.newBuilder(CONNECTION_ID).build();
 
     private ActorSystem actorSystem;
     private Connection connection;