@Test
public void testConditionalComparingAttributes() {
    AttributeSensor<String> sensor1 = Sensors.newStringSensor("sensor1");
    AttributeSensor<String> sensor2 = Sensors.newStringSensor("sensor2");
    TestEntity entity = app.createAndManageChild(EntitySpec.create(TestEntity.class));
    entity.sensors().set(sensor1, "myval1");
    entity.sensors().set(sensor2, "myval1");
    String templateContents = Joiner.on("\n").join(
            "[#ftl]",
            "[#if attribute['sensor1'] == attribute['sensor2']]",
            "true",
            "[#else]",
            "false",
            "[/#if]");
    String result = TemplateProcessor.processTemplateContents(templateContents, entity, ImmutableMap.<String,Object>of());
    assertEquals(result.trim(), "true");
    
    entity.sensors().set(sensor2, "myval2");
    String result2 = TemplateProcessor.processTemplateContents(templateContents, entity, ImmutableMap.<String,Object>of());
    assertEquals(result2.trim(), "false");
}
 

/**
 * The sensorVal must include port 1234, so that it will be converted to "+publicIp+":1234
 */
@Test(dataProvider = "variants")
public <T> void testSensorTransformed(Timing setUri, Timing addLocation, 
        AttributeSensor<T> sensor, T sensorVal, String targetSensorName, String expectedVal) throws Exception {
    entity.sensors().set(Attributes.SUBNET_ADDRESS, privateIp);
    if (setUri == Timing.BEFORE) {
        entity.sensors().set(sensor, sensorVal);
    }
    if (addLocation == Timing.BEFORE) {
        entity.addLocations(ImmutableList.of(machine));
    }
    
    entity.enrichers().add(EnricherSpec.create(OnSubnetNetworkEnricher.class)
            .configure(OnSubnetNetworkEnricher.SENSORS, ImmutableList.of(sensor)));

    if (setUri == Timing.AFTER) {
        entity.sensors().set(sensor, sensorVal);
    }
    if (addLocation == Timing.AFTER) {
        entity.addLocations(ImmutableList.of(machine));
    }
    
    EntityAsserts.assertAttributeEqualsEventually(entity, Sensors.newStringSensor(targetSensorName), expectedVal);
    EntityAsserts.assertAttributeEquals(entity, sensor, sensorVal);
}
 

@Override
public void onEvent(SensorEvent<Object> event) {
    Sensor<?> destinationSensor = getConfig(TARGET_SENSOR);

    List<Object> values = Lists.newArrayList();

    for (AttributeSensor<?> sourceSensor : subscribedSensors) {
        Object resolvedSensorValue = entity.sensors().get(sourceSensor);
        values.add(resolvedSensorValue);
    }
    Object result = reducerFunction.apply(values);

    if (LOG.isTraceEnabled()) LOG.trace("enricher {} got {}, propagating via {} as {}",
            new Object[] {this, event, entity, reducerFunction, destinationSensor});

    emit(destinationSensor, result);
}
 

@DataProvider(name = "variants")
public Object[][] provideVariants() {
    AttributeSensor<HostAndPort> hostAndPortSensor = Sensors.newSensor(HostAndPort.class, "test.endpoint");
    List<Object[]> result = Lists.newArrayList();
    for (Timing setSensor : Timing.values()) {
        for (Timing addLocation : Timing.values()) {
            result.add(new Object[] {setSensor, addLocation, Attributes.MAIN_URI, 
                    URI.create("http://"+publicIp+":1234/my/path"), "main.uri.mapped.subnet", "http://"+privateIp+":1234/my/path"});
            result.add(new Object[] {setSensor, addLocation, TestEntity.NAME, 
                    "http://"+publicIp+":1234/my/path", "test.name.mapped.subnet", "http://"+privateIp+":1234/my/path"});
            result.add(new Object[] {setSensor, addLocation, Attributes.HTTP_PORT, 
                    1234, "http.endpoint.mapped.subnet", privateIp+":1234"});
            result.add(new Object[] {setSensor, addLocation, TestEntity.NAME, 
                    "1234", "test.name.mapped.subnet", privateIp+":1234"});
            result.add(new Object[] {setSensor, addLocation, TestEntity.NAME, 
                    publicIp+":1234", "test.name.mapped.subnet", privateIp+":1234"});
            result.add(new Object[] {setSensor, addLocation, hostAndPortSensor, 
                    HostAndPort.fromString(publicIp+":1234"), "test.endpoint.mapped.subnet", privateIp+":1234"});
        }
    }
    return result.toArray(new Object[result.size()][]);
}
 

@Test
public void testReducingBuilderWithConcatenator() {
    entity.enrichers().add(Enrichers.builder()
            .reducing(Reducer.class, ImmutableList.<AttributeSensor<?>>of(STR1, STR2))
            .from(entity)
            .computing(new Concatenator())
            .publishing(STR3)
            .build()
    );

    EntityAsserts.assertAttributeEquals(entity, STR3, null);
    
    entity.sensors().set(STR1, "foo");
    EntityAsserts.assertAttributeEqualsContinually(entity, STR3, null);

    entity.sensors().set(STR2, "bar");
    EntityAsserts.assertAttributeEqualsEventually(entity, STR3, "foobar");
}
 

@Test
public void testAddsConditionalAndPropagatesSensors() throws Exception {
    optional = app.addChild(EntitySpec.create(ConditionalEntity.class)
            .configure(ConditionalEntity.CREATE_CONDITIONAL_ENTITY, true)
            .configure(ConditionalEntity.PROPAGATE_CONDITIONAL_ENTITY_SENSORS, true)
            .configure(ConditionalEntity.CONDITIONAL_ENTITY_SENSOR_LIST, ImmutableList.<AttributeSensor<?>>of(TestEntity.SEQUENCE))
            .configure(ConditionalEntity.CONDITIONAL_ENTITY_SPEC, EntitySpec.create(TestEntity.class)));
    app.start(ImmutableList.of(loc1));

    assertEquals(optional.getChildren().size(), 1);
    Entity child = Iterables.getOnlyElement(optional.getChildren());
    assertTrue(child instanceof TestEntity);
    assertEquals(child, optional.sensors().get(ConditionalEntity.CONDITIONAL_ENTITY));

    // Check that the configured sensors are propagated
    child.sensors().set(TestEntity.SEQUENCE, 123);
    EntityAsserts.assertAttributeEqualsEventually(child, TestEntity.SEQUENCE, 123);
    EntityAsserts.assertAttributeEqualsEventually(optional, TestEntity.SEQUENCE, 123);
    child.sensors().set(TestEntity.NAME, "frog");
    EntityAsserts.assertAttributeEqualsEventually(child, TestEntity.NAME, "frog");
    EntityAsserts.assertAttribute(optional, TestEntity.NAME, Predicates.isNull());
}
 

@Override
public Map<String, Object> batchSensorRead(final String application, final String entityToken, final Boolean raw) {
    final Entity entity = brooklyn().getEntity(application, entityToken);
    if (!Entitlements.isEntitled(mgmt().getEntitlementManager(), Entitlements.SEE_ENTITY, entity)) {
        throw WebResourceUtils.forbidden("User '%s' is not authorized to see entity '%s'",
                Entitlements.getEntitlementContext().user(), entity);
    }

    Map<String, Object> sensorMap = Maps.newHashMap();
    @SuppressWarnings("rawtypes")
    Iterable<AttributeSensor> sensors = filter(entity.getEntityType().getSensors(), AttributeSensor.class);

    for (AttributeSensor<?> sensor : sensors) {
        // Exclude sensors that user is not allowed to see
        if (!Entitlements.isEntitled(mgmt().getEntitlementManager(), Entitlements.SEE_SENSOR, new EntityAndItem<String>(entity, sensor.getName()))) {
            log.trace("User {} not authorized to see sensor {} of entity {}; excluding from current-state results", 
                    new Object[] {Entitlements.getEntitlementContext().user(), sensor.getName(), entity});
            continue;
        }

        Object value = entity.getAttribute(findSensor(entity, sensor.getName()));
        sensorMap.put(sensor.getName(), 
            resolving(value).preferJson(true).asJerseyOutermostReturnValue(false).raw(raw).context(entity).timeout(Duration.ZERO).renderAs(sensor).resolve());
    }
    return sensorMap;
}
 

@Test
public void testMonospaceTimeWeightedDeltaEnricher() {
    AttributeSensor<Double> deltaSensor = new BasicAttributeSensor<Double>(Double.class, "per second delta delta sensor");
    @SuppressWarnings("unchecked")
    TimeWeightedDeltaEnricher<Integer> delta = producer.enrichers().add(EnricherSpec.create(TimeWeightedDeltaEnricher.class)
            .configure("producer", producer)
            .configure("source", intSensor)
            .configure("target", deltaSensor)
            .configure("unitMillis", 1000));
    
    // Don't start with timestamp=0: that may be treated special 
    delta.onEvent(intSensor.newEvent(producer, 0), 1000);
    assertEquals(producer.getAttribute(deltaSensor), null);
    delta.onEvent(intSensor.newEvent(producer, 0), 2000);
    assertEquals(producer.getAttribute(deltaSensor), 0d);
    delta.onEvent(intSensor.newEvent(producer, 1), 3000);
    assertEquals(producer.getAttribute(deltaSensor), 1d);
    delta.onEvent(intSensor.newEvent(producer, 3), 4000);
    assertEquals(producer.getAttribute(deltaSensor), 2d);
    delta.onEvent(intSensor.newEvent(producer, 8), 5000);
    assertEquals(producer.getAttribute(deltaSensor), 5d);
}
 

@Test
public void testRegexReplacementFunctionWithAttributeWhenReady() throws Exception {
    Entity testEntity = setupAndCheckTestEntityInBasicYamlWith(
            "  brooklyn.enrichers:",
            "  - type: org.apache.brooklyn.enricher.stock.Transformer",
            "    brooklyn.config:",
            "      enricher.sourceSensor: $brooklyn:sensor(\"test.name\")",
            "      enricher.targetSensor: $brooklyn:sensor(\"test.name.transformed\")",
            "      enricher.transformation: $brooklyn:function.regexReplacement($brooklyn:attributeWhenReady(\"test.pattern\"), $brooklyn:attributeWhenReady(\"test.replacement\"))"
    );
    testEntity.sensors().set(Sensors.newStringSensor("test.pattern"), "foo");
    testEntity.sensors().set(Sensors.newStringSensor("test.replacement"), "bar");
    testEntity.sensors().set(TestEntity.NAME, "somefooname");
    AttributeSensor<String> transformedSensor = Sensors.newStringSensor("test.name.transformed");
    EntityAsserts.assertAttributeEqualsEventually(testEntity, transformedSensor, "somebarname");
}
 

@Test
public void testReducingBuilderWithJoinerFunction() {
    entity.enrichers().add(Enrichers.builder()
            .reducing(Reducer.class, ImmutableList.<AttributeSensor<?>>of(STR1, STR2))
            .from(entity)
            .computing("joiner", ImmutableMap.<String, Object>of("separator", "-"))
            .publishing(STR3)
            .build()
    );

    EntityAsserts.assertAttributeEquals(entity, STR3, null);
    
    entity.sensors().set(STR1, "foo");
    EntityAsserts.assertAttributeEqualsEventually(entity, STR3, "foo-null");

    entity.sensors().set(STR2, "bar");
    EntityAsserts.assertAttributeEqualsEventually(entity, STR3, "foo-bar");
}
 

@Test
public void testMetricResizingBeyondMaxMachines() throws Exception {
    AttributeSensor<Integer> metric = Sensors.newIntegerSensor("test.aggregatedLoad");
    
    cluster.resize(1);
    policy = cluster.policies().add(PolicySpec.create(AutoScalerPolicy.class)
            .configure(AutoScalerPolicy.METRIC, metric)
            .configure(AutoScalerPolicy.METRIC_LOWER_BOUND, 10)
            .configure(AutoScalerPolicy.METRIC_UPPER_BOUND, 20)
            .configure(AutoScalerPolicy.MIN_PERIOD_BETWEEN_EXECS, Duration.millis(10)));

    // Single node trying to handle a load of 21; too high, so will add one more node.
    // That takes the load back to within acceptable limits
    cluster.sensors().set(metric, 21);
    assertSizeEventually(2);
    cluster.sensors().set(metric, 19);

    // With two nodes, load is now too high, so will try (and fail) to add one more node.
    // Trigger another attempt to resize.
    // Any nodes that fail with NoMachinesAvailableException will be immediately deleted.
    cluster.sensors().set(metric, 22);
    assertSizeEventually(2, 0, 1);
    assertSizeContinually(2, 0, 1);
    
    // Metric is re-published; should not keep retrying
    cluster.sensors().set(metric, 21);
    assertSizeContinually(2, 0, 1);
}
 

/**
 * @deprecated since 0.11.0; explicit groovy utilities/support will be deleted.
 */
@Deprecated
public static <T,V> Task<V> attributePostProcessedWhenReady(Entity source, AttributeSensor<T> sensor, Closure<Boolean> ready, Closure<V> postProcess) {
    Predicate<? super T> readyPredicate = (ready != null) ? GroovyJavaMethods.predicateFromClosure(ready) : JavaGroovyEquivalents.groovyTruthPredicate();
    Function<? super T, V> postProcessFunction = GroovyJavaMethods.<T,V>functionFromClosure(postProcess);
    return attributePostProcessedWhenReady(source, sensor, readyPredicate, postProcessFunction);
}
 

@Test
public void testInteger() {
    AttributeSensor<Integer> currentSensor = new BasicAttributeSensor<Integer>(Integer.class, "current");
    AttributeSensor<Integer> totalSensor = new BasicAttributeSensor<Integer>(Integer.class, "total");

    app.enrichers().add(EnricherSpec.create(PercentageEnricher.class)
            .configure(PercentageEnricher.SOURCE_CURRENT_SENSOR, currentSensor)
            .configure(PercentageEnricher.SOURCE_TOTAL_SENSOR, totalSensor)
            .configure(PercentageEnricher.TARGET_SENSOR, targetSensor)
    );

    app.sensors().set(currentSensor, 25);
    app.sensors().set(totalSensor, 50);
    EntityAsserts.assertAttributeEqualsEventually(app, targetSensor, 0.5d);
}
 

/**
 * Deploy a cluster of 4 SoftwareProcesses, which all share the same CountingLatch instance so
 * that only two can obtain it at a time. The {@link CountingLatch} is configured with a really
 * long sleep after it acquires the lock, so we should have just 2 entities having acquired it
 * and the others blocked.
 * 
 * We assert that we got into this state, and then we tear it down by unmanaging the cluster 
 * (we unmanage because the cluster would otherwise takes ages due to the sleep in 
 * {@link CountingLatch}.
 */
@Test(dataProvider="latchAndTaskNamesProvider", timeOut=Asserts.THIRTY_SECONDS_TIMEOUT_MS)
public void testConcurrencyAllowsExactlyMax(ConfigKey<Boolean> latch, List<String> _ignored) throws Exception {
    boolean isLatchOnStop = latch.getName().contains("stop");
    LocalhostMachineProvisioningLocation loc = app.newLocalhostProvisioningLocation(ImmutableMap.of("address", "127.0.0.1"));
    
    final int maxConcurrency = 2;
    final ReleaseableLatch latchSemaphore = ReleaseableLatch.Factory.newMaxConcurrencyLatch(maxConcurrency);
    final AttributeSensor<Object> latchSensor = Sensors.newSensor(Object.class, "latch");
    final CountingLatch countingLatch = new CountingLatch(latchSemaphore, maxConcurrency, Asserts.DEFAULT_LONG_TIMEOUT.multiply(2));
    
    DynamicCluster cluster = app.createAndManageChild(EntitySpec.create(DynamicCluster.class)
            .configure(DynamicCluster.INITIAL_SIZE, maxConcurrency*2)
            .configure(DynamicCluster.MEMBER_SPEC, EntitySpec.create(MyService.class)
                    .configure(ConfigKeys.newConfigKey(Object.class, latch.getName()), (Object)DependentConfiguration.attributeWhenReady(app, latchSensor))));
    app.sensors().set(latchSensor, countingLatch);
    
    try {
        if (isLatchOnStop) {
            // Start will complete; then invoke stop async (don't expect it to complete!)
            app.start(ImmutableList.of(loc));
            Entities.invokeEffector(app, app, MyService.STOP, ImmutableMap.<String, Object>of());
        } else {
            // Invoke start async (don't expect it to complete!)
            Entities.invokeEffector(app, app, MyService.START, ImmutableMap.of("locations", ImmutableList.of(loc)));
        }
        
        // Because CountingLatch waits for ages, we'll eventually have maxConcurrent having successfully 
        // acquired, but the others blocked. Wait for that, and assert it stays that way.
        countingLatch.assertMaxCounterEventually(Predicates.equalTo(maxConcurrency));
        countingLatch.assertMaxCounterContinually(Predicates.equalTo(maxConcurrency), Duration.millis(100));
    } finally {
        // Don't wait for cluster to start/stop (because of big sleep in CountingLatch) - unmanage it.
        Entities.unmanage(cluster);
    }
}
 

/**
 * @deprecated since 0.12.0; use {@link EnricherSpec}
 */
@Deprecated
public RollingTimeWindowMeanEnricher(Entity producer, AttributeSensor<T> source, 
    AttributeSensor<Double> target, Duration timePeriod) {
    super(producer, source, target);
    this.timePeriod = Preconditions.checkNotNull(timePeriod, "timePeriod");
}
 

private static <T> void assertAttributeEqualsContinually(Entity x, AttributeSensor<T> sensor, T value) {
    try {
        EntityAsserts.assertAttributeEqualsContinually(ImmutableMap.of("timeout", Duration.millis(25)), x, sensor, value);
    } catch (Throwable e) {
        log.warn("Expected "+x+" continually to have "+sensor+" = "+value+"; instead:");
        Dumper.dumpInfo(x);
        throw Exceptions.propagate(e);
    }
}
 

/** returns an unsubmitted {@link Task} which blocks until the given sensor on the given source entity gives a value that satisfies ready, then returns that value;
 * particular useful in Entity configuration where config will block until Tasks have a value
 */
public static <T> Task<T> attributeWhenReady(final Entity source, final AttributeSensor<T> sensor, final Predicate<? super T> ready) {
    Builder<T, T> builder = builder().attributeWhenReady(source, sensor);
    if (ready != null) builder.readiness(ready);
    return builder.build();

}
 

@SuppressWarnings("deprecation")
@Test
public void testStoredByPathCanBeRetrieved() throws Exception {
    AttributeSensor<String> sensor1 = Sensors.newStringSensor("a", "");
    AttributeSensor<String> sensor2 = Sensors.newStringSensor("b.c", "");

    map.update(ImmutableList.of("a"), "1val");
    map.update(ImmutableList.of("b", "c"), "2val");
    
    assertEquals(map.getValue(sensor1), "1val");
    assertEquals(map.getValue(sensor2), "2val");
    
    assertEquals(map.getValue(ImmutableList.of("a")), "1val");
    assertEquals(map.getValue(ImmutableList.of("b","c")), "2val");
}
 

protected Collection<AttributeSensor<?>> resolveSensorsConfig() {
    Collection<? extends AttributeSensor<?>> sensors = getConfig(SENSORS);

    Collection<AttributeSensor<?>> result = Lists.newArrayList();
    if (sensors != null) {
        for (Object s : sensors) {
            AttributeSensor<?> coercedSensor = TypeCoercions.coerce(s, AttributeSensor.class);
            AttributeSensor<?> typedSensor = (AttributeSensor<?>) entity.getEntityType().getSensor(coercedSensor.getName());
            result.add(typedSensor != null ? typedSensor : coercedSensor);
        }
    }
    return result;
}
 

public static <T> void assertAttributeEqualsEventually(Map<?,?> flags, final Entity entity, final AttributeSensor<T> attribute, final T expected) {
    // Not using assertAttributeEventually(predicate) so get nicer error message
    Asserts.succeedsEventually(castToMapWithStringKeys(flags), new Runnable() {
        @Override
        public void run() {
            assertAttributeEquals(entity, attribute, expected);
        }
    });
}
 

@Test
public void testDoNotRepersistOnSetAttributeWithSameValue() throws Exception {
    final AttributeSensor<String> MY_ATTRIBUTE = Sensors.builder(String.class, "myAttribute").build();
    
    origApp.sensors().set(MY_ATTRIBUTE, "myval");
    RebindTestUtils.waitForPersisted(mgmt());
    
    origApp.sensors().set(MY_ATTRIBUTE, "myval");
    assertFalse(RebindTestUtils.hasPendingPersists(mgmt()));
}
 

protected <T> SubscriptionHandle recordEvents(Entity entity, AttributeSensor<T> sensor, final List<SensorEvent<T>> events) {
    SensorEventListener<T> listener = new SensorEventListener<T>() {
        @Override public void onEvent(SensorEvent<T> event) {
            log.info("onEvent: {}", event);
            events.add(event);
        }
    };
    return entity.subscriptions().subscribe(entity, sensor, listener);
}
 

@Test(groups="Integration")
public void testResizeFailsWhenEntitlementThrowsShouldLogException() throws Exception {
    final AttributeSensor<Integer> scalingMetric = Sensors.newIntegerSensor("scalingMetric");
    
    ControlledDynamicWebAppCluster cluster = app.createAndManageChild(EntitySpec.create(ControlledDynamicWebAppCluster.class)
            .configure("initialSize", 1)
            .configure(ControlledDynamicWebAppCluster.CONTROLLER_SPEC, EntitySpec.create(TrackingAbstractController.class))
            .configure(ControlledDynamicWebAppCluster.MEMBER_SPEC, EntitySpec.create(TestJavaWebAppEntity.class))
            .policy(PolicySpec.create(AutoScalerPolicy.class)
                    .configure(AutoScalerPolicy.METRIC, scalingMetric)
                    .configure(AutoScalerPolicy.MIN_POOL_SIZE, 1)
                    .configure(AutoScalerPolicy.MAX_POOL_SIZE, 2)
                    .configure(AutoScalerPolicy.METRIC_LOWER_BOUND, 1)
                    .configure(AutoScalerPolicy.METRIC_UPPER_BOUND, 10)));
    
    Entitlements.setEntitlementContext(new EntitlementContext() {
            @Override
            public String user() {
                return "myuser";
            }});
    app.start(ImmutableList.of(loc));
    Entitlements.clearEntitlementContext();

    String loggerName = EffectorUtils.class.getName();
    ch.qos.logback.classic.Level logLevel = ch.qos.logback.classic.Level.DEBUG;
    Predicate<ILoggingEvent> filter = Predicates.and(EventPredicates.containsMessage("Error invoking "), 
            EventPredicates.containsExceptionStackLine(ThrowingEntitlementManager.class, "isEntitled"));
    LogWatcher watcher = new LogWatcher(loggerName, logLevel, filter);
            
    watcher.start();
    try {
        // Cause the auto-scaler to resize, which will fail because of the bad entitlement implementation
        cluster.sensors().set(scalingMetric, 50);
        watcher.assertHasEventEventually();
    } finally {
        watcher.close();
    }
}
 

@Test
public void testPropagatorDefaultsToProducerAsSelf() throws Exception {
    AttributeSensor<String> sourceSensor = Sensors.newSensor(String.class, "mySensor");
    AttributeSensor<String> targetSensor = Sensors.newSensor(String.class, "myTarget");

    app.enrichers().add(EnricherSpec.create(Propagator.class)
            .configure(Propagator.PRODUCER, app)
            .configure(Propagator.SENSOR_MAPPING, ImmutableMap.of(sourceSensor, targetSensor)));

    app.sensors().set(sourceSensor, "myval");
    EntityAsserts.assertAttributeEqualsEventually(app, targetSensor, "myval");
}
 

@Override
public void onException(Exception exception) {
    log.error("Detected exception while retrieving Chef attributes from entity " + entity.getDisplayName(), exception);
    for (AttributeSensor<?> attribute : chefAttributeSensors.values()) {
        if (!attribute.getName().startsWith(CHEF_ATTRIBUTE_PREFIX))
            continue;
        entity.sensors().set(attribute, null);
    }
}
 

public Task<V2> build() {
    List<Task<V>> tasks = MutableList.of();
    for (AttributeAndSensorCondition<?> source: multiSource) {
        builder.source(source.source);
        builder.sensor((AttributeSensor)source.sensor);
        builder.readiness((Predicate)source.predicate);
        tasks.add(builder.build());
    }
    final Task<List<V>> parallelTask = Tasks.<List<V>>builder().parallel(true).addAll(tasks)
        .displayName(name)
        .description(descriptionBase+
            (builder.timeout!=null ? ", timeout "+builder.timeout : ""))
        .build();
    
    if (postProcessFromMultiple == null) {
        // V2 should be the right type in normal operations
        return (Task<V2>) parallelTask;
    } else {
        return Tasks.<V2>builder().displayName(name).description(descriptionBase)
            .tag("attributeWhenReady")
            .body(new Callable<V2>() {
                @Override public V2 call() throws Exception {
                    List<V> prePostProgress = DynamicTasks.queue(parallelTask).get();
                    return DynamicTasks.queue(
                        Tasks.<V2>builder().displayName("post-processing").description("Applying "+postProcessFromMultiple)
                            .body(Functionals.callable(postProcessFromMultiple, prePostProgress))
                            .build()).get();
                }
            })
            .build();
    }
}
 

@Test
public void testLong() {
    AttributeSensor<Long> currentSensor = new BasicAttributeSensor<Long>(Long.class, "current");
    AttributeSensor<Long> totalSensor = new BasicAttributeSensor<Long>(Long.class, "total");

    app.enrichers().add(EnricherSpec.create(PercentageEnricher.class)
            .configure(PercentageEnricher.SOURCE_CURRENT_SENSOR, currentSensor)
            .configure(PercentageEnricher.SOURCE_TOTAL_SENSOR, totalSensor)
            .configure(PercentageEnricher.TARGET_SENSOR, targetSensor)
    );

    app.sensors().set(currentSensor, 25l);
    app.sensors().set(totalSensor, 50l);
    EntityAsserts.assertAttributeEqualsEventually(app, targetSensor, 0.5d);
}
 

protected void registerPortMappings(KubernetesSshMachineLocation machine, Entity entity, Service service) {
    PortForwardManager portForwardManager = (PortForwardManager) getManagementContext().getLocationRegistry()
            .getLocationManaged(PortForwardManagerLocationResolver.PFM_GLOBAL_SPEC);
    List<ServicePort> ports = service.getSpec().getPorts();
    String publicHostText = machine.getSshHostAndPort().getHostText();
    LOG.debug("Recording port-mappings for container {} of {}: {}", machine, this, ports);

    for (ServicePort port : ports) {
        String protocol = port.getProtocol();
        Integer targetPort = port.getTargetPort().getIntVal();

        if (!"TCP".equalsIgnoreCase(protocol)) {
            LOG.debug("Ignoring port mapping {} for {} because only TCP is currently supported", port, machine);
        } else if (targetPort == null) {
            LOG.debug("Ignoring port mapping {} for {} because targetPort.intValue is null", port, machine);
        } else if (port.getNodePort() == null) {
            LOG.debug("Ignoring port mapping {} to {} because port.getNodePort() is null", targetPort, machine);
        } else {
            portForwardManager.associate(publicHostText, HostAndPort.fromParts(publicHostText, port.getNodePort()), machine, targetPort);
            AttributeSensor<Integer> sensor = Sensors.newIntegerSensor("kubernetes." + Strings.maybeNonBlank(port.getName()).or(targetPort.toString()) + ".port");
            entity.sensors().set(sensor, targetPort);
        }
    }

    entity.enrichers().add(EnricherSpec.create(OnPublicNetworkEnricher.class)
            .configure(AbstractOnNetworkEnricher.MAP_MATCHING, "kubernetes.[a-zA-Z0-9][a-zA-Z0-9-_]*.port"));
}
 

@Test(dataProvider="latchAndTaskNamesProvider"/*, timeOut=Asserts.THIRTY_SECONDS_TIMEOUT_MS*/)
public void testFailedReleaseableUnblocks(final ConfigKey<Boolean> latch, List<String> _ignored) throws Exception {
    LocalhostMachineProvisioningLocation loc = app.newLocalhostProvisioningLocation(ImmutableMap.of("address", "127.0.0.1"));

    final int maxConcurrency = 1;
    final ReleaseableLatch latchSemaphore = ReleaseableLatch.Factory.newMaxConcurrencyLatch(maxConcurrency);
    final AttributeSensor<Object> latchSensor = Sensors.newSensor(Object.class, "latch");
    final CountingLatch countingLatch = new CountingLatch(latchSemaphore, maxConcurrency);
    // FIRST_MEMBER_SPEC latches are not guaranteed to be acquired before MEMBER_SPEC latches
    // so the start effector could complete, but the counting latch will catch if there are
    // any unreleased semaphores.
    @SuppressWarnings({"unused"})
    DynamicCluster cluster = app.createAndManageChild(EntitySpec.create(DynamicCluster.class)
            .configure(DynamicCluster.INITIAL_SIZE, 2)
            .configure(DynamicCluster.FIRST_MEMBER_SPEC, EntitySpec.create(FailingMyService.class)
                    .configure(ConfigKeys.newConfigKey(Object.class, latch.getName()), (Object)DependentConfiguration.attributeWhenReady(app, latchSensor)))
            .configure(DynamicCluster.MEMBER_SPEC, EntitySpec.create(MyService.class)
                    .configure(ConfigKeys.newConfigKey(Object.class, latch.getName()), (Object)DependentConfiguration.attributeWhenReady(app, latchSensor))));
    app.sensors().set(latchSensor, countingLatch);
    final Task<Void> startTask = Entities.invokeEffector(app, app, MyService.START, ImmutableMap.of("locations", ImmutableList.of(loc)));
    //expected to fail but should complete quickly
    assertTrue(startTask.blockUntilEnded(Asserts.DEFAULT_LONG_TIMEOUT), "timeout waiting for start effector to complete");
    assertTrue(latch == SoftwareProcess.STOP_LATCH || startTask.isError());
    final Task<Void> stopTask = Entities.invokeEffector(app, app, MyService.STOP, ImmutableMap.<String, Object>of());
    //expected to fail but should complete quickly
    assertTrue(stopTask.blockUntilEnded(Asserts.DEFAULT_LONG_TIMEOUT), "timeout waiting for stop effector to complete");
    // stop task won't fail because the process stop failed; the error is ignored
    assertTrue(stopTask.isDone());
    assertEquals(countingLatch.getCounter(), 0);
    // Check we have actually used the latch
    assertNotEquals(countingLatch.getMaxCounter(), 0, "Latch not acquired at all");
    // In theory this is 0 < maxCnt <= maxConcurrency contract, but in practice
    // we should always reach the maximum due to the sleeps in CountingLatch.
    // Change if found to fail in the wild.
    assertEquals(countingLatch.getMaxCounter(), maxConcurrency);
}
 

private void testTransformerResolvesResolvableValues(int portStart, int portCount) {
    // Note: The test gets progressively slower with iterations, probably due to the GC triggering much more frequently.
    //       There's no memory leak, but doesn't seem right to be putting so much pressure on the GC with such a simple test.
    AttributeSensor<Integer> sourceSensor = Sensors.newIntegerSensor("port");
    AttributeSensor<String> targetSensor = Sensors.newStringSensor("port.transformed");
    app.enrichers().add(EnricherSpec.create(Transformer.class)
            .configure(Transformer.SOURCE_SENSOR, sourceSensor)
            .configure(Transformer.TARGET_SENSOR, targetSensor)
            .configure(Transformer.TARGET_VALUE,
                    // Can only use the inner-most sensor, but including the
                    // wrapping formatStrings amplifies the resolving effort, making
                    // a bug more probable to manifest.
                    BrooklynDslCommon.formatString("%s",
                            BrooklynDslCommon.formatString("%d",
                                    BrooklynDslCommon.attributeWhenReady("port")))));

    int failures = 0;
    for (int port = portStart; port < portStart + portCount; port++) {
        app.sensors().set(sourceSensor, port);
        try {
            EntityAsserts.assertAttributeEqualsEventually(app, targetSensor, Integer.toString(port));
        } catch (Exception e) {
            failures++;
            LOG.warn("Assertion failed, port=" + port + ", transformed sensor is " + app.sensors().get(targetSensor), e);
        }
    }

    assertEquals(failures, 0, failures + " assertion failures while transforming sensor; see logs for detailed errors");
}