fix RedisPubSubTest

This test used to fail intermittently. The problem is that when the `SUBSCRIBE`
command returns, there's no guarantee that the subscription has actually
completed. That is only guaranteed after the subscription connection receives
the `subscribe` message. The fix is to simply wait until that happens and
only then run `PUBLISH` on the publish connection.

Author: Ladicek

src/test/java/io/vertx/tests/redis/client/RedisPubSubTest.java

@@ -18,13 +18,14 @@
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
-import java.util.Set;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static io.vertx.redis.client.Command.PSUBSCRIBE;
 import static io.vertx.redis.client.Command.PUBLISH;
 import static io.vertx.redis.client.Command.SUBSCRIBE;
 import static io.vertx.redis.client.Request.cmd;
+import static io.vertx.tests.redis.client.TestUtils.executeWhenConditionSatisfied;
 
 @RunWith(VertxUnitRunner.class)
 public class RedisPubSubTest {
@@ -73,8 +74,12 @@ public void publishSubscribe_withHandler(TestContext test) {
     rule.vertx().eventBus().consumer("io.vertx.redis.news", msg -> async.countDown());
 
     subConn.handler(EventBusHandler.create(rule.vertx()));
-    subConn.send(Request.cmd(Command.SUBSCRIBE).arg("news")).onComplete(test.asyncAssertSuccess(result -> {
-      pubConn.send(Request.cmd(Command.PUBLISH).arg("news").arg("foo"));
+    subConn.send(Request.cmd(Command.SUBSCRIBE).arg("news")).onComplete(test.asyncAssertSuccess(subscribe -> {
+      executeWhenConditionSatisfied(rule.vertx(), () -> async.count() == 1, () -> {
+        pubConn.send(Request.cmd(Command.PUBLISH).arg("news").arg("foo")).onComplete(test.asyncAssertSuccess(publish -> {
+          test.assertEquals(1, publish.toInteger());
+        }));
+      });
     }));
   }
 
@@ -85,8 +90,12 @@ public void publishPSubscribe_withHandler(TestContext test) {
     rule.vertx().eventBus().consumer("io.vertx.redis.new*", msg -> async.countDown());
 
     subConn.handler(EventBusHandler.create(rule.vertx()));
-    subConn.send(Request.cmd(Command.PSUBSCRIBE).arg("new*")).onComplete(test.asyncAssertSuccess(result -> {
-      pubConn.send(Request.cmd(Command.PUBLISH).arg("news").arg("foo"));
+    subConn.send(Request.cmd(Command.PSUBSCRIBE).arg("new*")).onComplete(test.asyncAssertSuccess(subscribe -> {
+      executeWhenConditionSatisfied(rule.vertx(), () -> async.count() == 1, () -> {
+        pubConn.send(Request.cmd(Command.PUBLISH).arg("news").arg("foo")).onComplete(test.asyncAssertSuccess(publish -> {
+          test.assertEquals(1, publish.toInteger());
+        }));
+      });
     }));
   }
 
@@ -119,8 +128,11 @@ public void publishSubscribe_naive(TestContext test) {
     });
 
     subConn.send(cmd(SUBSCRIBE).arg("mychannel")).onComplete(test.asyncAssertSuccess(subscribe -> {
-      pubConn.send(cmd(PUBLISH).arg("mychannel").arg(123456))
-        .onComplete(test.asyncAssertSuccess(test::assertNotNull));
+      executeWhenConditionSatisfied(rule.vertx(), () -> subscribeCnt.get() == 1, () -> {
+        pubConn.send(cmd(PUBLISH).arg("mychannel").arg(123456)).onComplete(test.asyncAssertSuccess(publish -> {
+          test.assertEquals(1, publish.toInteger());
+        }));
+      });
     }));
   }
 
@@ -152,15 +164,19 @@ public void publishPSubscribe_naive(TestContext test) {
       }
     });
 
-    Set<String> patterns = Set.of("A*", "B*", "C*", "D*", "E*", "F*");
+    List<String> patterns = List.of("A*", "B*", "C*", "D*", "E*", "F*");
+    List<String> matchingChannels = List.of("A", "B1", "Co", "DDD", "E234", "F");
 
     Request psub = cmd(PSUBSCRIBE);
     patterns.forEach(psub::arg);
 
     subConn.send(psub).onComplete(test.asyncAssertSuccess(subscribe -> {
-      patterns.forEach(p -> {
-        pubConn.send(cmd(PUBLISH).arg(p).arg(System.nanoTime()))
-          .onComplete(test.asyncAssertSuccess(test::assertNotNull));
+      executeWhenConditionSatisfied(rule.vertx(), () -> psubscribeCnt.get() == 6, () -> {
+        for (String ch : matchingChannels) {
+          pubConn.send(cmd(PUBLISH).arg(ch).arg(System.nanoTime())).onComplete(test.asyncAssertSuccess(publish -> {
+            test.assertEquals(1, publish.toInteger());
+          }));
+        }
       });
     }));
   }

src/test/java/io/vertx/tests/redis/client/TestUtils.java

@@ -2,17 +2,44 @@
 
 import io.vertx.core.Completable;
 import io.vertx.core.Future;
+import io.vertx.core.Handler;
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 
 import java.util.UUID;
+import java.util.function.BooleanSupplier;
 import java.util.function.Supplier;
 
 public class TestUtils {
+  /**
+   * Returns a short random string that is usable as a key in Redis with very high probability
+   * of being unique.
+   * <p>
+   * Implementation note: this is currently a random UUID, but that is <em>not</em> guaranteed
+   * and may change without notice.
+   */
   public static String randomKey() {
     return UUID.randomUUID().toString();
   }
 
+  /**
+   * Waits until the given {@code condition} is {@code true} and then invokes the given {@code action} once,
+   * returning its result. Waiting is not active, it uses the Vert.x {@linkplain Vertx#setTimer(long, Handler) timer}
+   * facility.
+   */
+  public static void executeWhenConditionSatisfied(Vertx vertx, BooleanSupplier condition, Runnable action) {
+    if (condition.getAsBoolean()) {
+      action.run();
+    } else {
+      vertx.setTimer(5, ignored -> {
+        executeWhenConditionSatisfied(vertx, condition, action);
+      });
+    }
+  }
+
+  /**
+   * Retries the given {@code action} until it succeeds or until the number of retries reaches the given maximum.
+   */
   public static <T> Future<T> retryUntilSuccess(Vertx vertx, Supplier<Future<T>> action, int maxRetries) {
     Promise<T> promise = Promise.promise();
     retryUntilSuccess(vertx, action, maxRetries, promise);