Backport 584 to 0.10.x

core/src/main/java/org/apache/brooklyn/core/entity/Entities.java

@@ -18,6 +18,8 @@
  */
 package org.apache.brooklyn.core.entity;
 
+import static org.apache.brooklyn.util.guava.Functionals.isSatisfied;
+
 import java.io.Closeable;
 import java.io.IOException;
 import java.io.PrintWriter;
@@ -32,7 +34,6 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.Stack;
-import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Semaphore;
@@ -1232,24 +1233,32 @@ public static void warnOnIgnoringConfig(Entity entity, ConfigKey<?> key) {
             log.warn("Ignoring "+key+" set on "+entity+" ("+entity.getConfig(key)+")");
     }
 
-    /** Waits until {@link Startable#SERVICE_UP} returns true. */
-    public static void waitForServiceUp(final Entity entity, Duration timeout) {
-        String description = "Waiting for SERVICE_UP on "+entity;
-        Tasks.setBlockingDetails(description);
+    /** Waits until the passed entity satisfies the supplied predicate. */
+    public static void waitFor(Entity entity, Predicate<Entity> condition, Duration timeout) {
         try {
-            if (!Repeater.create(description).limitTimeTo(timeout)
-                    .rethrowException().backoffTo(Duration.ONE_SECOND)
-                    .until(new Callable<Boolean>() {
-                        public Boolean call() {
-                            return Boolean.TRUE.equals(entity.getAttribute(Startable.SERVICE_UP));
-                        }})
-                    .run()) {
-                throw new IllegalStateException("Timeout waiting for SERVICE_UP from "+entity);
+            String description = "Waiting for " + condition + " on " + entity;
+            Tasks.setBlockingDetails(description);
+
+            Repeater repeater = Repeater.create(description)
+                .until(isSatisfied(entity, condition))
+                .limitTimeTo(timeout)
+                .backoffTo(Duration.ONE_SECOND)
+                .rethrowException();
+
+            if (!repeater.run()) {
+                throw new IllegalStateException("Timeout waiting for " + condition + " on " + entity);
             }
+
         } finally {
             Tasks.resetBlockingDetails();
         }
-        log.debug("Detected SERVICE_UP for software {}", entity);
+
+        log.debug("Detected {} for {}", condition, entity);
+    }
+
+    /** Waits until {@link Startable#SERVICE_UP} is true. */
+    public static void waitForServiceUp(final Entity entity, Duration timeout) {
+        waitFor(entity, EntityPredicates.isServiceUp(), timeout);
     }
     public static void waitForServiceUp(final Entity entity, long duration, TimeUnit units) {
         waitForServiceUp(entity, Duration.of(duration, units));

core/src/main/java/org/apache/brooklyn/core/entity/EntityPredicates.java

@@ -30,6 +30,7 @@
 import org.apache.brooklyn.config.ConfigKey;
 import org.apache.brooklyn.config.ConfigKey.HasConfigKey;
 import org.apache.brooklyn.core.config.ConfigKeys;
+import org.apache.brooklyn.core.entity.trait.Startable;
 import org.apache.brooklyn.core.sensor.Sensors;
 import org.apache.brooklyn.util.collections.CollectionFunctionals;
 import org.apache.brooklyn.util.guava.SerializablePredicate;
@@ -502,4 +503,20 @@ public boolean apply(@Nullable Entity input) {
         };
     }
 
+    public static Predicate<Entity> isServiceUp() {
+        return new IsServiceUp();
+    }
+
+    /** Common test, provide short friendly toString(). */
+    protected static class IsServiceUp implements SerializablePredicate<Entity> {
+        @Override
+        public boolean apply(Entity input) {
+            return Boolean.TRUE.equals(input.sensors().get(Startable.SERVICE_UP));
+        }
+        @Override
+        public String toString() {
+            return "SERVICE_UP";
+        }
+    };
+
 }

core/src/test/java/org/apache/brooklyn/core/entity/EntitiesTest.java

@@ -18,6 +18,7 @@
  */
 package org.apache.brooklyn.core.entity;
 
+import static org.apache.brooklyn.core.entity.EntityPredicates.applicationIdEqualTo;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
@@ -26,14 +27,12 @@
 
 import org.apache.brooklyn.api.entity.EntitySpec;
 import org.apache.brooklyn.api.location.LocationSpec;
-import org.apache.brooklyn.core.entity.Entities;
-import org.apache.brooklyn.core.entity.EntityAndAttribute;
-import org.apache.brooklyn.core.entity.EntityInitializers;
 import org.apache.brooklyn.core.location.SimulatedLocation;
 import org.apache.brooklyn.core.test.BrooklynAppUnitTestSupport;
 import org.apache.brooklyn.core.test.entity.TestEntity;
 import org.apache.brooklyn.test.Asserts;
 import org.apache.brooklyn.util.collections.MutableSet;
+import org.apache.brooklyn.util.time.Duration;
 import org.testng.Assert;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
@@ -171,4 +170,19 @@ public void testCreateGetContainsAndRemoveTags() throws Exception {
         Assert.assertEquals(entity.tags().getTags(), MutableSet.of(app));
     }
 
+    @Test
+    public void testWaitFor() throws Exception {
+        entity = app.createAndManageChild(EntitySpec.create(TestEntity.class));
+        Duration timeout = Duration.ONE_MILLISECOND;
+
+        Entities.waitFor(entity, applicationIdEqualTo(app.getApplicationId()), timeout);
+
+        try {
+            Entities.waitFor(entity, applicationIdEqualTo(app.getApplicationId() + "-wrong"), timeout);
+            Asserts.shouldHaveFailedPreviously("Entities.waitFor() should have timed out");
+        } catch (Exception e) {
+            Asserts.expectedFailureContains(e, "Timeout waiting for ");
+        }
+    }
+
 }

utils/common/src/main/java/org/apache/brooklyn/util/guava/Functionals.java

@@ -148,4 +148,16 @@ public String toString() {
         return new FunctionAsPredicate();
     }
 
+    /**
+     * Simple adapter from {@link Predicate} to {@link Callable} by currying the passed <tt>subject</tt> parameter.
+     */
+    public static <T> Callable<Boolean> isSatisfied(final T subject, final Predicate<T> predicate) {
+        return new Callable<Boolean>() {
+            @Override
+            public Boolean call() {
+                return predicate.apply(subject);
+            }
+        };
+    }
+
 }

utils/common/src/test/java/org/apache/brooklyn/util/guava/FunctionalsTest.java

@@ -18,11 +18,11 @@
  */
 package org.apache.brooklyn.util.guava;
 
-import org.apache.brooklyn.util.guava.Functionals;
 import org.apache.brooklyn.util.math.MathFunctions;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
+import com.google.common.base.Predicate;
 import com.google.common.base.Predicates;
 import com.google.common.base.Suppliers;
 
@@ -55,4 +55,15 @@ public void testIfNotEqual() {
         IfFunctionsTest.checkTF(Functionals.ifNotEquals(false).value("T").defaultValue("F").build(), "T");
     }
 
+    @Test
+    public void testIsSatisfied() throws Exception {
+        Predicate<Integer> isEven = new Predicate<Integer>() {
+            @Override public boolean apply(Integer input) {
+                return (input % 2 == 0);
+            }
+        };
+        Assert.assertFalse(Functionals.isSatisfied(11, isEven).call());
+        Assert.assertTrue(Functionals.isSatisfied(22, isEven).call());
+    }
+
 }