AbstractInvokeEffectorPolicy extends AbstractPolicy { protected <T> Task<T> invoke(Effector<T> effector, Map<String, ?> parameters) { if (isBusySensorEnabled()) { getTaskCounter().incrementAndGet(); publishIsBusy(); } Task<T> task = entity.invoke(effector, parameters); if (isBusySensorEnabled()) { task.addListener(new EffectorListener(), MoreExecutors.sameThreadExecutor()); } return task; }  @Override void setEntity(EntityLocal entity); static final ConfigKey<String> IS_BUSY_SENSOR_NAME; }
@Test public void testCountReflectsNumberOfExecutingEffectors() { final CountDownLatch effectorLatch = new CountDownLatch(1); final AttributeSensor<Boolean> policyIsBusy = Sensors.newBooleanSensor( "policyIsBusy"); final Effector<Void> blockingEffector = Effectors.effector(Void.class, "abstract-invoke-effector-policy-test") .impl(new BlockingEffector(effectorLatch)) .build(); final TestEntity entity = app.createAndManageChild(EntitySpec.create(TestEntity.class)); final TestAbstractInvokeEffectorPolicy policy = entity.policies().add( PolicySpec.create(TestAbstractInvokeEffectorPolicy.class) .configure(AbstractInvokeEffectorPolicy.IS_BUSY_SENSOR_NAME, policyIsBusy.getName())); final Task<?> effectorTask = policy.invoke(blockingEffector, ImmutableMap.<String, Object>of()); Supplier<Boolean> effectorTaskDoneSupplier = new Supplier<Boolean>() { @Override public Boolean get() { return effectorTask.isDone(); } }; Asserts.continually(effectorTaskDoneSupplier, Predicates.equalTo(false)); EntityAsserts.assertAttributeEqualsEventually(entity, policyIsBusy, true); effectorLatch.countDown(); Asserts.eventually(effectorTaskDoneSupplier, Predicates.equalTo(true)); EntityAsserts.assertAttributeEqualsEventually(entity, policyIsBusy, false); }