Fix issues with rebind
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/0a71cf58 Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/0a71cf58 Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/0a71cf58 Branch: refs/heads/master Commit: 0a71cf58b018cde15b2f71f8b97dc063b871ebd6 Parents: 00deb7d Author: Andrew Donald Kennedy <[email protected]> Authored: Fri Sep 15 17:14:11 2017 +0100 Committer: Andrew Donald Kennedy <[email protected]> Committed: Fri Sep 15 17:21:19 2017 +0100 ---------------------------------------------------------------------- .../action/AbstractScheduledEffectorPolicy.java | 114 ++++++++++++------- .../policy/action/PeriodicEffectorPolicy.java | 49 +++++--- .../policy/action/ScheduledEffectorPolicy.java | 72 ++---------- .../action/PeriodicEffectorPolicyTest.java | 18 ++- .../action/ScheduledEffectorPolicyTest.java | 26 ++--- 5 files changed, 139 insertions(+), 140 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/0a71cf58/policy/src/main/java/org/apache/brooklyn/policy/action/AbstractScheduledEffectorPolicy.java ---------------------------------------------------------------------- diff --git a/policy/src/main/java/org/apache/brooklyn/policy/action/AbstractScheduledEffectorPolicy.java b/policy/src/main/java/org/apache/brooklyn/policy/action/AbstractScheduledEffectorPolicy.java index 73438c6..5978ec5 100644 --- a/policy/src/main/java/org/apache/brooklyn/policy/action/AbstractScheduledEffectorPolicy.java +++ b/policy/src/main/java/org/apache/brooklyn/policy/action/AbstractScheduledEffectorPolicy.java @@ -23,9 +23,11 @@ import java.text.ParseException; import java.text.SimpleDateFormat; import java.util.Calendar; import java.util.Date; +import java.util.List; import java.util.Map; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.brooklyn.api.effector.Effector; @@ -38,11 +40,11 @@ import org.apache.brooklyn.core.config.ConfigKeys; import org.apache.brooklyn.core.entity.EntityInitializers; import org.apache.brooklyn.core.entity.trait.Startable; import org.apache.brooklyn.core.policy.AbstractPolicy; -import org.apache.brooklyn.util.collections.MutableMap; import org.apache.brooklyn.util.core.config.ConfigBag; import org.apache.brooklyn.util.core.config.ResolvingConfigBag; import org.apache.brooklyn.util.core.task.Tasks; import org.apache.brooklyn.util.exceptions.Exceptions; +import org.apache.brooklyn.util.exceptions.RuntimeInterruptedException; import org.apache.brooklyn.util.guava.Maybe; import org.apache.brooklyn.util.time.Duration; import org.apache.brooklyn.util.time.DurationPredicates; @@ -53,6 +55,7 @@ import com.google.common.annotations.Beta; import com.google.common.base.Predicates; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Iterables; +import com.google.common.collect.Lists; import com.google.common.reflect.TypeToken; @Beta @@ -103,20 +106,35 @@ public abstract class AbstractScheduledEffectorPolicy extends AbstractPolicy imp .reconfigurable(true) .build(); - protected final AtomicBoolean running = new AtomicBoolean(false); - protected final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); - protected final Object mutex = new Object[0]; + public static final ConfigKey<List<Long>> SCHEDULED = ConfigKeys.builder(new TypeToken<List<Long>>() { }) + .name("scheduled") + .description("List of all scheduled execution start times") + .defaultValue(Lists.newCopyOnWriteArrayList()) + .reconfigurable(true) + .build(); - protected Effector<?> effector; + protected AtomicBoolean running; + protected ScheduledExecutorService executor; + protected Effector effector; public AbstractScheduledEffectorPolicy() { - this(MutableMap.<String,Object>of()); + LOG.debug("Created new scheduled effector policy"); + } + + @Override + public void init() { + setup(); } - public AbstractScheduledEffectorPolicy(Map<String,?> props) { - super(props); + public void setup() { + if (executor != null) { + executor.shutdownNow(); + } + executor = Executors.newSingleThreadScheduledExecutor(); + running = new AtomicBoolean(false); } + @Override public void setEntity(EntityLocal entity) { super.setEntity(entity); @@ -128,9 +146,22 @@ public abstract class AbstractScheduledEffectorPolicy extends AbstractPolicy imp @Override public void rebind() { + setup(); + if (config().get(RUNNING)) { running.set(true); - start(); + } + + if (running.get()) { + List<Long> scheduled = config().get(SCHEDULED); + for (Long when : scheduled) { + Duration wait = Duration.millis(when - System.currentTimeMillis()); + if (wait.isPositive()) { + schedule(wait); + } else { + scheduled.remove(when); + } + } } } @@ -145,15 +176,15 @@ public abstract class AbstractScheduledEffectorPolicy extends AbstractPolicy imp @Override public void destroy(){ - super.destroy(); executor.shutdownNow(); + super.destroy(); } public abstract void start(); protected Effector<?> getEffector() { String effectorName = config().get(EFFECTOR); - Maybe<Effector<?>> effector = entity.getEntityType().getEffectorByName(effectorName); + Maybe<Effector<?>> effector = getEntity().getEntityType().getEffectorByName(effectorName); if (effector.isAbsentOrNull()) { throw new IllegalStateException("Cannot find effector " + effectorName); } @@ -181,39 +212,46 @@ public abstract class AbstractScheduledEffectorPolicy extends AbstractPolicy imp } } + protected void schedule(Duration wait) { + List<Long> scheduled = config().get(SCHEDULED); + scheduled.add(System.currentTimeMillis() + wait.toMilliseconds()); + + executor.schedule(this, wait.toMilliseconds(), TimeUnit.MILLISECONDS); + } + @Override - public void run() { - synchronized (mutex) { - try { - ConfigBag bag = ResolvingConfigBag.newInstanceExtending(getManagementContext(), config().getBag()); - Map<String, Object> args = EntityInitializers.resolve(bag, EFFECTOR_ARGUMENTS); - LOG.debug("{}: Resolving arguments for {}: {}", new Object[] { this, effector.getName(), Iterables.toString(args.keySet()) }); - Map<String, Object> resolved = (Map) Tasks.resolving(args, Object.class) - .deep(true) - .context(entity) - .get(); - - LOG.debug("{}: Invoking effector on {}, {}({})", new Object[] { this, entity, effector.getName(), resolved }); - Object result = entity.invoke(effector, resolved).getUnchecked(); - LOG.debug("{}: Effector {} returned {}", new Object[] { this, effector.getName(), result }); - } catch (Throwable t) { - LOG.warn("{}: Exception running {}: {}", new Object[] { this, effector.getName(), t.getMessage() }); - Exceptions.propagate(t); - } + public synchronized void run() { + if (effector == null) return; + try { + ConfigBag bag = ResolvingConfigBag.newInstanceExtending(getManagementContext(), config().getBag()); + Map<String, Object> args = EntityInitializers.resolve(bag, EFFECTOR_ARGUMENTS); + LOG.debug("{}: Resolving arguments for {}: {}", new Object[] { this, effector.getName(), Iterables.toString(args.keySet()) }); + Map<String, Object> resolved = (Map) Tasks.resolving(args, Object.class) + .deep(true) + .context(entity) + .get(); + + LOG.debug("{}: Invoking effector on {}, {}({})", new Object[] { this, entity, effector.getName(), resolved }); + Object result = entity.invoke(effector, resolved).getUnchecked(); + LOG.debug("{}: Effector {} returned {}", new Object[] { this, effector.getName(), result }); + } catch (RuntimeInterruptedException rie) { + Thread.interrupted(); + // TODO sometimes this seems to hang the executor? + } catch (Throwable t) { + LOG.warn("{}: Exception running {}: {}", new Object[] { this, effector.getName(), t.getMessage() }); + Exceptions.propagate(t); } } @Override public void onEvent(SensorEvent<Object> event) { - synchronized (mutex) { - LOG.debug("{}: Got event {}", this, event); - AttributeSensor<Boolean> sensor = config().get(START_SENSOR); - if (event.getSensor().getName().equals(sensor.getName())) { - Boolean start = (Boolean) event.getValue(); - if (start && running.compareAndSet(false, true)) { - config().set(RUNNING, true); - start(); - } + LOG.debug("{}: Got event {}", this, event); + AttributeSensor<Boolean> sensor = config().get(START_SENSOR); + if (event.getSensor().getName().equals(sensor.getName())) { + Boolean start = (Boolean) event.getValue(); + if (start && running.compareAndSet(false, true)) { + config().set(RUNNING, true); + start(); } } } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/0a71cf58/policy/src/main/java/org/apache/brooklyn/policy/action/PeriodicEffectorPolicy.java ---------------------------------------------------------------------- diff --git a/policy/src/main/java/org/apache/brooklyn/policy/action/PeriodicEffectorPolicy.java b/policy/src/main/java/org/apache/brooklyn/policy/action/PeriodicEffectorPolicy.java index 58c10f2..f2c0936 100644 --- a/policy/src/main/java/org/apache/brooklyn/policy/action/PeriodicEffectorPolicy.java +++ b/policy/src/main/java/org/apache/brooklyn/policy/action/PeriodicEffectorPolicy.java @@ -18,18 +18,17 @@ */ package org.apache.brooklyn.policy.action; -import java.util.Map; -import java.util.concurrent.TimeUnit; +import java.util.List; import org.apache.brooklyn.api.effector.Effector; import org.apache.brooklyn.api.entity.EntityLocal; import org.apache.brooklyn.api.policy.Policy; import org.apache.brooklyn.config.ConfigKey; import org.apache.brooklyn.core.config.ConfigKeys; -import org.apache.brooklyn.util.collections.MutableMap; +import org.apache.brooklyn.util.exceptions.Exceptions; +import org.apache.brooklyn.util.text.Strings; import org.apache.brooklyn.util.time.Duration; import org.apache.brooklyn.util.time.DurationPredicates; -import org.apache.brooklyn.util.time.Time; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -72,16 +71,7 @@ public class PeriodicEffectorPolicy extends AbstractScheduledEffectorPolicy { .build(); public PeriodicEffectorPolicy() { - this(MutableMap.<String,Object>of()); - } - - public PeriodicEffectorPolicy(Map<String,?> props) { - super(props); - } - - @Override - public void setEntity(final EntityLocal entity) { - super.setEntity(entity); + super(); } @Override @@ -95,8 +85,33 @@ public class PeriodicEffectorPolicy extends AbstractScheduledEffectorPolicy { wait = period; } - LOG.debug("{}: Scheduling {} every {} in {}", new Object[] { PeriodicEffectorPolicy.this, effector.getName(), - Time.fromDurationToTimeStringRounded().apply(period), Time.fromDurationToTimeStringRounded().apply(wait) }); - executor.scheduleAtFixedRate(PeriodicEffectorPolicy.this, wait.toMilliseconds(), period.toMilliseconds(), TimeUnit.MILLISECONDS); + schedule(wait); + } + + @Override + public void rebind() { + super.rebind(); + + // Check if we missed an entire period + List<Long> scheduled = config().get(SCHEDULED); + if (running.get() && scheduled.isEmpty()) { + start(); + } + } + + @Override + public synchronized void run() { + try { + super.run(); + } finally { + Duration period = config().get(PERIOD); + String time = config().get(TIME); + if (time == null || time.equalsIgnoreCase(NOW) || time.equalsIgnoreCase(IMMEDIATELY)) { + schedule(period); + } else { + Duration wait = getWaitUntil(time); + schedule(wait.upperBound(period)); + } + } } } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/0a71cf58/policy/src/main/java/org/apache/brooklyn/policy/action/ScheduledEffectorPolicy.java ---------------------------------------------------------------------- diff --git a/policy/src/main/java/org/apache/brooklyn/policy/action/ScheduledEffectorPolicy.java b/policy/src/main/java/org/apache/brooklyn/policy/action/ScheduledEffectorPolicy.java index 44e6aac..5a3bed7 100644 --- a/policy/src/main/java/org/apache/brooklyn/policy/action/ScheduledEffectorPolicy.java +++ b/policy/src/main/java/org/apache/brooklyn/policy/action/ScheduledEffectorPolicy.java @@ -19,27 +19,18 @@ package org.apache.brooklyn.policy.action; import java.util.Date; -import java.util.List; -import java.util.Map; -import java.util.concurrent.TimeUnit; import org.apache.brooklyn.api.effector.Effector; import org.apache.brooklyn.api.entity.EntityLocal; import org.apache.brooklyn.api.policy.Policy; import org.apache.brooklyn.api.sensor.AttributeSensor; import org.apache.brooklyn.api.sensor.SensorEvent; -import org.apache.brooklyn.config.ConfigKey; -import org.apache.brooklyn.core.config.ConfigKeys; import org.apache.brooklyn.core.sensor.Sensors; -import org.apache.brooklyn.util.collections.MutableMap; import org.apache.brooklyn.util.time.Duration; -import org.apache.brooklyn.util.time.Time; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.google.common.annotations.Beta; -import com.google.common.collect.Lists; -import com.google.common.reflect.TypeToken; /** * A {@link Policy} the executes an {@link Effector} at a specific time in the future. @@ -57,22 +48,11 @@ public class ScheduledEffectorPolicy extends AbstractScheduledEffectorPolicy { private static final Logger LOG = LoggerFactory.getLogger(ScheduledEffectorPolicy.class); - public static final ConfigKey<List<Long>> SCHEDULED = ConfigKeys.builder(new TypeToken<List<Long>>() { }) - .name("scheduled") - .description("List of all scheduled execution start times") - .defaultValue(Lists.newCopyOnWriteArrayList()) - .reconfigurable(true) - .build(); - public static final AttributeSensor<Boolean> INVOKE_IMMEDIATELY = Sensors.newBooleanSensor("scheduler.invoke.now", "Invoke the configured effector immediately when this becomes true"); public static final AttributeSensor<Date> INVOKE_AT = Sensors.newSensor(Date.class, "scheduler.invoke.at", "Invoke the configured effector at this time"); public ScheduledEffectorPolicy() { - this(MutableMap.<String,Object>of()); - } - - public ScheduledEffectorPolicy(Map<String,?> props) { - super(props); + super(); } @Override @@ -84,27 +64,11 @@ public class ScheduledEffectorPolicy extends AbstractScheduledEffectorPolicy { } @Override - public void rebind() { - super.rebind(); - List<Long> scheduled = config().get(SCHEDULED); - for (Long when : scheduled) { - Duration wait = Duration.millis(when - System.currentTimeMillis()); - if (wait.isPositive()) { - schedule(wait); - } else { - scheduled.remove(when); - } - } - } - - @Override public void start() { String time = config().get(TIME); Duration wait = config().get(WAIT); if (time != null) { - LOG.debug("{}: Scheduling {} at {} (in {})", - new Object[] { this, effector.getName(), time, Time.fromDurationToTimeStringRounded().apply(wait) }); wait = getWaitUntil(time); } @@ -113,35 +77,23 @@ public class ScheduledEffectorPolicy extends AbstractScheduledEffectorPolicy { } } - protected void schedule(Duration wait) { - List<Long> scheduled = config().get(SCHEDULED); - scheduled.add(System.currentTimeMillis() + wait.toMilliseconds()); - - LOG.debug("{}: Scheduling {} in {} ({} ms)", - new Object[] { this, effector.getName(), Time.fromDurationToTimeStringRounded().apply(wait), wait.toMilliseconds() }); - executor.schedule(this, wait.toMilliseconds(), TimeUnit.MILLISECONDS); - } - @Override public void onEvent(SensorEvent<Object> event) { - synchronized (mutex) { - super.onEvent(event); + super.onEvent(event); - if (running.get()) { - if (event.getSensor().getName().equals(INVOKE_AT.getName())) { - String time = (String) event.getValue(); - if (time != null) { - schedule(getWaitUntil(time)); - } + if (running.get()) { + if (event.getSensor().getName().equals(INVOKE_AT.getName())) { + String time = (String) event.getValue(); + if (time != null) { + schedule(getWaitUntil(time)); } - if (event.getSensor().getName().equals(INVOKE_IMMEDIATELY.getName())) { - Boolean invoke = (Boolean) event.getValue(); - if (invoke) { - schedule(Duration.ZERO); - } + } + if (event.getSensor().getName().equals(INVOKE_IMMEDIATELY.getName())) { + Boolean invoke = (Boolean) event.getValue(); + if (invoke) { + schedule(Duration.ZERO); } } } } - } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/0a71cf58/policy/src/test/java/org/apache/brooklyn/policy/action/PeriodicEffectorPolicyTest.java ---------------------------------------------------------------------- diff --git a/policy/src/test/java/org/apache/brooklyn/policy/action/PeriodicEffectorPolicyTest.java b/policy/src/test/java/org/apache/brooklyn/policy/action/PeriodicEffectorPolicyTest.java index e6ae74d..0268d60 100644 --- a/policy/src/test/java/org/apache/brooklyn/policy/action/PeriodicEffectorPolicyTest.java +++ b/policy/src/test/java/org/apache/brooklyn/policy/action/PeriodicEffectorPolicyTest.java @@ -37,24 +37,24 @@ import com.google.common.collect.Iterables; public class PeriodicEffectorPolicyTest extends BrooklynAppUnitTestSupport { + private static final AttributeSensor<Boolean> START = Sensors.newBooleanSensor("start"); + @Test public void testPeriodicEffectorFires() { - final AttributeSensor<Boolean> start = Sensors.newBooleanSensor("start"); - - final TestEntity entity = app.createAndManageChild(EntitySpec.create(TestEntity.class) + TestEntity entity = app.createAndManageChild(EntitySpec.create(TestEntity.class) .policy(PolicySpec.create(PeriodicEffectorPolicy.class) .configure(PeriodicEffectorPolicy.EFFECTOR, "myEffector") .configure(PeriodicEffectorPolicy.EFFECTOR_ARGUMENTS, ImmutableMap.of()) .configure(PeriodicEffectorPolicy.PERIOD, Duration.ONE_MILLISECOND) .configure(PeriodicEffectorPolicy.TIME, "immediately") - .configure(PeriodicEffectorPolicy.START_SENSOR, start))); + .configure(PeriodicEffectorPolicy.START_SENSOR, START))); Policy policy = Iterables.tryFind(entity.policies(), Predicates.instanceOf(PeriodicEffectorPolicy.class)).orNull(); Asserts.assertNotNull(policy); Asserts.assertTrue(entity.getCallHistory().isEmpty()); Asserts.assertFalse(policy.config().get(PeriodicEffectorPolicy.RUNNING)); - entity.sensors().set(start, Boolean.TRUE); + entity.sensors().set(START, Boolean.TRUE); Asserts.eventually(() -> policy.config().get(PeriodicEffectorPolicy.RUNNING), b -> b); Asserts.eventually(() -> entity.getCallHistory(), l -> l.contains("myEffector")); int calls = entity.getCallHistory().size(); @@ -63,22 +63,20 @@ public class PeriodicEffectorPolicyTest extends BrooklynAppUnitTestSupport { @Test public void testPeriodicEffectorFiresAfterDelay() { - final AttributeSensor<Boolean> start = Sensors.newBooleanSensor("start"); - - final TestEntity entity = app.createAndManageChild(EntitySpec.create(TestEntity.class) + TestEntity entity = app.createAndManageChild(EntitySpec.create(TestEntity.class) .policy(PolicySpec.create(PeriodicEffectorPolicy.class) .configure(PeriodicEffectorPolicy.EFFECTOR, "myEffector") .configure(PeriodicEffectorPolicy.EFFECTOR_ARGUMENTS, ImmutableMap.of()) .configure(PeriodicEffectorPolicy.PERIOD, Duration.ONE_MILLISECOND) .configure(PeriodicEffectorPolicy.WAIT, Duration.TEN_SECONDS) - .configure(PeriodicEffectorPolicy.START_SENSOR, start))); + .configure(PeriodicEffectorPolicy.START_SENSOR, START))); Policy policy = Iterables.tryFind(entity.policies(), Predicates.instanceOf(PeriodicEffectorPolicy.class)).orNull(); Asserts.assertNotNull(policy); Asserts.assertTrue(entity.getCallHistory().isEmpty()); Asserts.assertFalse(policy.config().get(PeriodicEffectorPolicy.RUNNING)); - entity.sensors().set(start, Boolean.TRUE); + entity.sensors().set(START, Boolean.TRUE); Asserts.eventually(() -> policy.config().get(PeriodicEffectorPolicy.RUNNING), b -> b); sleep(Duration.seconds(5)); Asserts.eventually(() -> entity.getCallHistory(), l -> !l.contains("myEffector")); http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/0a71cf58/policy/src/test/java/org/apache/brooklyn/policy/action/ScheduledEffectorPolicyTest.java ---------------------------------------------------------------------- diff --git a/policy/src/test/java/org/apache/brooklyn/policy/action/ScheduledEffectorPolicyTest.java b/policy/src/test/java/org/apache/brooklyn/policy/action/ScheduledEffectorPolicyTest.java index daa186e..5271de3 100644 --- a/policy/src/test/java/org/apache/brooklyn/policy/action/ScheduledEffectorPolicyTest.java +++ b/policy/src/test/java/org/apache/brooklyn/policy/action/ScheduledEffectorPolicyTest.java @@ -37,44 +37,42 @@ import com.google.common.collect.Iterables; public class ScheduledEffectorPolicyTest extends BrooklynAppUnitTestSupport { + private static final AttributeSensor<Boolean> START = Sensors.newBooleanSensor("start"); + @Test public void testScheduledEffectorFiresImmediately() { - final AttributeSensor<Boolean> start = Sensors.newBooleanSensor("start"); - - final TestEntity entity = app.createAndManageChild(EntitySpec.create(TestEntity.class) + TestEntity entity = app.createAndManageChild(EntitySpec.create(TestEntity.class) .policy(PolicySpec.create(ScheduledEffectorPolicy.class) .configure(ScheduledEffectorPolicy.EFFECTOR, "myEffector") .configure(ScheduledEffectorPolicy.EFFECTOR_ARGUMENTS, ImmutableMap.of()) .configure(ScheduledEffectorPolicy.TIME, "immediately") - .configure(PeriodicEffectorPolicy.START_SENSOR, start))); + .configure(PeriodicEffectorPolicy.START_SENSOR, START))); Policy policy = Iterables.tryFind(entity.policies(), Predicates.instanceOf(ScheduledEffectorPolicy.class)).orNull(); Asserts.assertNotNull(policy); Asserts.assertTrue(entity.getCallHistory().isEmpty()); Asserts.assertFalse(policy.config().get(ScheduledEffectorPolicy.RUNNING)); - entity.sensors().set(start, Boolean.TRUE); + entity.sensors().set(START, Boolean.TRUE); Asserts.eventually(() -> policy.config().get(ScheduledEffectorPolicy.RUNNING), b -> b); Asserts.eventually(() -> entity.getCallHistory(), l -> l.contains("myEffector")); } @Test public void testScheduledEffectorFiresAfterDelay() { - final AttributeSensor<Boolean> start = Sensors.newBooleanSensor("start"); - - final TestEntity entity = app.createAndManageChild(EntitySpec.create(TestEntity.class) + TestEntity entity = app.createAndManageChild(EntitySpec.create(TestEntity.class) .policy(PolicySpec.create(ScheduledEffectorPolicy.class) .configure(ScheduledEffectorPolicy.EFFECTOR, "myEffector") .configure(ScheduledEffectorPolicy.EFFECTOR_ARGUMENTS, ImmutableMap.of()) .configure(ScheduledEffectorPolicy.WAIT, Duration.TEN_SECONDS) - .configure(ScheduledEffectorPolicy.START_SENSOR, start))); + .configure(ScheduledEffectorPolicy.START_SENSOR, START))); Policy policy = Iterables.tryFind(entity.policies(), Predicates.instanceOf(ScheduledEffectorPolicy.class)).orNull(); Asserts.assertNotNull(policy); Asserts.assertTrue(entity.getCallHistory().isEmpty()); Asserts.assertFalse(policy.config().get(ScheduledEffectorPolicy.RUNNING)); - entity.sensors().set(start, Boolean.TRUE); + entity.sensors().set(START, Boolean.TRUE); Asserts.eventually(() -> policy.config().get(ScheduledEffectorPolicy.RUNNING), b -> b); sleep(Duration.seconds(5)); Asserts.eventually(() -> entity.getCallHistory(), l -> !l.contains("myEffector")); @@ -84,20 +82,18 @@ public class ScheduledEffectorPolicyTest extends BrooklynAppUnitTestSupport { @Test public void testScheduledEffectorFiresOnSensor() { - final AttributeSensor<Boolean> start = Sensors.newBooleanSensor("start"); - - final TestEntity entity = app.createAndManageChild(EntitySpec.create(TestEntity.class) + TestEntity entity = app.createAndManageChild(EntitySpec.create(TestEntity.class) .policy(PolicySpec.create(ScheduledEffectorPolicy.class) .configure(ScheduledEffectorPolicy.EFFECTOR, "myEffector") .configure(ScheduledEffectorPolicy.EFFECTOR_ARGUMENTS, ImmutableMap.of()) - .configure(ScheduledEffectorPolicy.START_SENSOR, start))); + .configure(ScheduledEffectorPolicy.START_SENSOR, START))); Policy policy = Iterables.tryFind(entity.policies(), Predicates.instanceOf(ScheduledEffectorPolicy.class)).orNull(); Asserts.assertNotNull(policy); Asserts.assertTrue(entity.getCallHistory().isEmpty()); Asserts.assertFalse(policy.config().get(ScheduledEffectorPolicy.RUNNING)); - entity.sensors().set(start, Boolean.TRUE); + entity.sensors().set(START, Boolean.TRUE); Asserts.eventually(() -> policy.config().get(ScheduledEffectorPolicy.RUNNING), b -> b); sleep(Duration.seconds(5)); Asserts.eventually(() -> entity.getCallHistory(), l -> !l.contains("myEffector"));
