Remove test intermittency by waiting on egress publishers to be conne…

…cted
lob-software · Apr 9, 2022 · 62802d3 · 62802d3
1 parent 83571ac
commit 62802d3
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 3 deletions.
diff --git a/aeronic-processor/src/main/java/io/aeronic/cluster/AeronicClusteredServiceContainer.java b/aeronic-processor/src/main/java/io/aeronic/cluster/AeronicClusteredServiceContainer.java
@@ -37,6 +37,11 @@ public <T> T getPublisherFor(final Class<T> clazz)
         return registry.getPublisherFor(clazz);
     }
 
+    public boolean egressConnected()
+    {
+        return registry.egressConnected();
+    }
+
     public static class Configuration
     {
         private ClusteredService clusteredService;

diff --git a/aeronic-processor/src/main/java/io/aeronic/cluster/AeronicClusteredServiceRegistry.java b/aeronic-processor/src/main/java/io/aeronic/cluster/AeronicClusteredServiceRegistry.java
@@ -39,6 +39,11 @@ public void registerIngressSubscriberInvoker(final AbstractSubscriberInvoker<?>
             .forEach(subscriberInterface -> invokerByName.put(subscriberInterface + "__IngressPublisher", subscriberInvoker));
     }
 
+    public boolean egressConnected()
+    {
+        return clientSessionPublicationByName.values().stream().allMatch(ClientSessionPublication::isConnected);
+    }
+
     public void onSessionOpen(final ClientSession session)
     {
         final byte[] encodedPrincipal = session.encodedPrincipal();

diff --git a/aeronic-tests/src/test/java/io/aeronic/system/cluster/ClusterSystemTest.java b/aeronic-tests/src/test/java/io/aeronic/system/cluster/ClusterSystemTest.java
@@ -72,15 +72,15 @@ public void clusterIngress()
             .clusteredService(clusteredService)
             .registerIngressSubscriber(SimpleEvents.class, simpleEvents)
             .registerIngressSubscriber(SampleEvents.class, sampleEvents)
-            .registerEgressPublisher(SimpleEvents.class)
-            .registerEgressPublisher(SampleEvents.class)
             .create();
 
-        clusterNode = new TestClusterNode(aeronicClusteredService,true);
+        clusterNode = new TestClusterNode(aeronicClusteredService, true);
 
         final SimpleEvents simpleEventsPublisher = aeronic.createClusterIngressPublisher(SimpleEvents.class, INGRESS_CHANNEL);
         final SampleEvents sampleEventsPublisher = aeronic.createClusterIngressPublisher(SampleEvents.class, INGRESS_CHANNEL);
 
+        aeronic.awaitUntilPubsAndSubsConnect();
+
         simpleEventsPublisher.onEvent(101L);
         sampleEventsPublisher.onEvent(201L);
 
@@ -108,6 +108,8 @@ public void clusterEgress()
         aeronic.registerClusterEgressSubscriber(SimpleEvents.class, simpleEvents, INGRESS_CHANNEL);
         aeronic.registerClusterEgressSubscriber(SampleEvents.class, sampleEvents, INGRESS_CHANNEL);
         aeronic.start();
+        aeronic.awaitUntilPubsAndSubsConnect();
+        await().timeout(Duration.ofSeconds(1)).until(aeronicClusteredService::egressConnected);
 
         simpleEventsPublisher.onEvent(101L);
         sampleEventsPublisher.onEvent(202L);
@@ -144,6 +146,8 @@ public void clusterIngressAndEgress()
         aeronic.registerClusterEgressSubscriber(SimpleEvents.class, simpleEvents, INGRESS_CHANNEL);
         aeronic.registerClusterEgressSubscriber(SampleEvents.class, sampleEvents, INGRESS_CHANNEL);
         aeronic.start();
+        aeronic.awaitUntilPubsAndSubsConnect();
+        await().timeout(Duration.ofSeconds(1)).until(aeronicClusteredService::egressConnected);
 
         // cluster -> client
         clusterEgressSimpleEventsPublisher.onEvent(101L);