You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@brooklyn.apache.org by he...@apache.org on 2016/08/25 12:10:37 UTC
[2/5] brooklyn-server git commit: Update Aggregator and Joiner
entichers to handle Map sensors
Update Aggregator and Joiner entichers to handle Map sensors
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/c58fe29e
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/c58fe29e
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/c58fe29e
Branch: refs/heads/master
Commit: c58fe29e3f44baf4e647aec1e3147edfbdddfd87
Parents: e8cce9b
Author: Andrew Donald Kennedy <an...@cloudsoftcorp.com>
Authored: Sat Aug 20 15:05:33 2016 +0100
Committer: Andrew Donald Kennedy <an...@cloudsoftcorp.com>
Committed: Sat Aug 20 17:09:12 2016 +0100
----------------------------------------------------------------------
.../stock/AbstractMultipleSensorAggregator.java | 14 ++--
.../brooklyn/enricher/stock/Enrichers.java | 64 +++++++++------
.../apache/brooklyn/enricher/stock/Joiner.java | 40 +++++++---
.../brooklyn/enricher/stock/MapAggregator.java | 83 ++++++++++++++++++++
.../brooklyn/enricher/stock/EnrichersTest.java | 40 ++++++++++
5 files changed, 199 insertions(+), 42 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/c58fe29e/core/src/main/java/org/apache/brooklyn/enricher/stock/AbstractMultipleSensorAggregator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/enricher/stock/AbstractMultipleSensorAggregator.java b/core/src/main/java/org/apache/brooklyn/enricher/stock/AbstractMultipleSensorAggregator.java
index e0f0d6d..c0cd36d 100644
--- a/core/src/main/java/org/apache/brooklyn/enricher/stock/AbstractMultipleSensorAggregator.java
+++ b/core/src/main/java/org/apache/brooklyn/enricher/stock/AbstractMultipleSensorAggregator.java
@@ -42,20 +42,19 @@ public abstract class AbstractMultipleSensorAggregator<U> extends AbstractAggreg
private static final Logger LOG = LoggerFactory.getLogger(AbstractMultipleSensorAggregator.class);
-
/** access via {@link #getValues(Sensor)} */
private final Map<String, Map<Entity,Object>> values = Collections.synchronizedMap(new LinkedHashMap<String, Map<Entity,Object>>());
- public AbstractMultipleSensorAggregator() {}
+ public AbstractMultipleSensorAggregator() { }
protected abstract Collection<Sensor<?>> getSourceSensors();
-
+
@Override
protected void setEntityLoadingConfig() {
super.setEntityLoadingConfig();
Preconditions.checkNotNull(getSourceSensors(), "sourceSensors must be set");
}
-
+
@Override
protected void setEntityBeforeSubscribingProducerChildrenEvents() {
BrooklynLogging.log(LOG, BrooklynLogging.levelDebugOrTraceIfReadOnly(producer),
@@ -95,7 +94,7 @@ public abstract class AbstractMultipleSensorAggregator<U> extends AbstractAggreg
vs = new LinkedHashMap<Entity,Object>();
values.put(sensor.getName(), vs);
}
-
+
Object vo = vs.get(producer);
if (vo==null) {
Object initialVal;
@@ -107,11 +106,10 @@ public abstract class AbstractMultipleSensorAggregator<U> extends AbstractAggreg
vs.put(producer, initialVal != null ? initialVal : defaultMemberValue);
// NB: see notes on possible race, in Aggregator#onProducerAdded
}
-
}
}
}
-
+
@Override
protected void onProducerRemoved(Entity producer) {
synchronized (values) {
@@ -163,7 +161,7 @@ public abstract class AbstractMultipleSensorAggregator<U> extends AbstractAggreg
return MutableMap.copyOf(sv).asUnmodifiable();
}
}
-
+
@Override
protected abstract Object compute();
}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/c58fe29e/core/src/main/java/org/apache/brooklyn/enricher/stock/Enrichers.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/enricher/stock/Enrichers.java b/core/src/main/java/org/apache/brooklyn/enricher/stock/Enrichers.java
index 3dfabcf..973933c 100644
--- a/core/src/main/java/org/apache/brooklyn/enricher/stock/Enrichers.java
+++ b/core/src/main/java/org/apache/brooklyn/enricher/stock/Enrichers.java
@@ -161,7 +161,11 @@ public class Enrichers {
}
/** as {@link #combining(Collection)} but the collection of values comes from the given sensor on multiple entities */
public <S> AggregatorBuilder<S, Object> aggregating(AttributeSensor<S> val) {
- return new AggregatorBuilder<S,Object>(val);
+ return new AggregatorBuilder<S, Object>(val);
+ }
+ /** as {@link #combining(Collection)} but the collection of values comes from the given sensor on multiple entities */
+ public <S> AggregatorBuilder<S, Object> aggregating(AttributeSensor<S> keySensor, AttributeSensor<S> valueSensor) {
+ return new AggregatorBuilder<S, Object>(keySensor, valueSensor);
}
/** creates an {@link UpdatingMap} enricher:
* {@link UpdatingMapBuilder#from(AttributeSensor)} and {@link UpdatingMapBuilder#computing(Function)} are required
@@ -182,13 +186,15 @@ public class Enrichers {
protected abstract static class AbstractAggregatorBuilder<S, T, B extends AbstractAggregatorBuilder<S, T, B>> extends AbstractEnricherBuilder<B> {
- protected final AttributeSensor<S> aggregating;
+ protected AttributeSensor<S> aggregating;
+ protected AttributeSensor<S> keySensor;
+ protected AttributeSensor<S> valueSensor;
protected AttributeSensor<T> publishing;
protected Entity fromEntity;
/** @deprecated since 0.7.0, kept for backwards compatibility for rebind, but not used otherwise */
@Deprecated protected Function<? super Collection<S>, ? extends T> computing;
// use supplier so latest values of other fields can be used
- protected Supplier<Function<? super Collection<S>, ? extends T>> computingSupplier;
+ protected Supplier<Function<? super Collection<S>, ? extends T>> computingSupplier = Suppliers.ofInstance(null);
protected Boolean fromMembers;
protected Boolean fromChildren;
protected Boolean excludingBlank;
@@ -197,11 +203,16 @@ public class Enrichers {
protected Predicate<Object> valueFilter;
protected Object defaultValueForUnreportedSensors;
protected Object valueToReportIfNoSensors;
-
+
public AbstractAggregatorBuilder(AttributeSensor<S> aggregating) {
super(Aggregator.class);
this.aggregating = aggregating;
}
+ public AbstractAggregatorBuilder(AttributeSensor<S> keySensor, AttributeSensor<S> valueSensor) {
+ super(MapAggregator.class);
+ this.keySensor = keySensor;
+ this.valueSensor = valueSensor;
+ }
@SuppressWarnings({ "unchecked", "rawtypes" })
public <T2 extends T> AggregatorBuilder<S,T2> publishing(AttributeSensor<? extends T2> val) {
this.publishing = (AttributeSensor) checkNotNull(val);
@@ -259,7 +270,6 @@ public class Enrichers {
};
return self();
}
-
public B computingAverage() {
this.computingSupplier = new Supplier<Function<? super Collection<S>, ? extends T>>() {
@Override
@@ -283,6 +293,10 @@ public class Enrichers {
this.entityFilter = val;
return self();
}
+ public B valueFilter(Predicate<Object> val) {
+ this.valueFilter = val;
+ return self();
+ }
public B excludingBlank() {
this.excludingBlank = true;
return self();
@@ -293,42 +307,33 @@ public class Enrichers {
return "aggregator:"+publishing.getName();
}
public EnricherSpec<?> build() {
- Predicate<Object> valueFilter;
- if (Boolean.TRUE.equals(excludingBlank)) {
- valueFilter = new Predicate<Object>() {
- @Override public boolean apply(Object input) {
- return (input != null) &&
- ((input instanceof CharSequence) ? Strings.isNonBlank((CharSequence)input) : true);
- }
- };
- // above kept for deserialization; not sure necessary
- valueFilter = StringPredicates.isNonBlank();
- } else {
- valueFilter = null;
- }
- // FIXME excludingBlank; use valueFilter? exclude means ignored entirely or substituted for defaultMemberValue?
return super.build().configure(MutableMap.builder()
.putIfNotNull(Aggregator.PRODUCER, fromEntity)
.put(Aggregator.TARGET_SENSOR, publishing)
- .put(Aggregator.SOURCE_SENSOR, aggregating)
+ .putIfNotNull(Aggregator.SOURCE_SENSOR, aggregating)
+ .putIfNotNull(MapAggregator.KEY_SENSOR, keySensor)
+ .putIfNotNull(MapAggregator.VALUE_SENSOR, valueSensor)
.putIfNotNull(Aggregator.FROM_CHILDREN, fromChildren)
.putIfNotNull(Aggregator.FROM_MEMBERS, fromMembers)
.putIfNotNull(Aggregator.TRANSFORMATION, computingSupplier.get())
.putIfNotNull(Aggregator.FROM_HARDCODED_PRODUCERS, fromHardcodedProducers)
+ .putIfNotNull(Aggregator.EXCLUDE_BLANK, excludingBlank)
.putIfNotNull(Aggregator.ENTITY_FILTER, entityFilter)
.putIfNotNull(Aggregator.VALUE_FILTER, valueFilter)
.putIfNotNull(Aggregator.DEFAULT_MEMBER_VALUE, defaultValueForUnreportedSensors)
.build());
}
-
+
@Override
public String toString() {
return Objects.toStringHelper(this)
.omitNullValues()
.add("aggregating", aggregating)
+ .add("keySensor", keySensor)
+ .add("valueSensor", valueSensor)
.add("publishing", publishing)
.add("fromEntity", fromEntity)
- .add("computing", computingSupplier)
+ .add("computing", computingSupplier.get())
.add("fromMembers", fromMembers)
.add("fromChildren", fromChildren)
.add("excludingBlank", excludingBlank)
@@ -638,6 +643,8 @@ public class Enrichers {
protected AttributeSensor<String> publishing;
protected Entity fromEntity;
protected String separator;
+ protected String keyValueSeparator;
+ protected Boolean joinMapEntries;
protected Boolean quote;
protected Integer minimum;
protected Integer maximum;
@@ -654,6 +661,14 @@ public class Enrichers {
this.separator = separator;
return self();
}
+ public B keyValueSeparator(String keyValueSeparator) {
+ this.keyValueSeparator = keyValueSeparator;
+ return self();
+ }
+ public B joinMapEntries(Boolean joinMapEntries) {
+ this.joinMapEntries = joinMapEntries;
+ return self();
+ }
public B quote(Boolean quote) {
this.quote = quote;
return self();
@@ -677,6 +692,8 @@ public class Enrichers {
.put(Joiner.TARGET_SENSOR, publishing)
.put(Joiner.SOURCE_SENSOR, transforming)
.putIfNotNull(Joiner.SEPARATOR, separator)
+ .putIfNotNull(Joiner.KEY_VALUE_SEPARATOR, keyValueSeparator)
+ .putIfNotNull(Joiner.JOIN_MAP_ENTRIES, joinMapEntries)
.putIfNotNull(Joiner.QUOTE, quote)
.putIfNotNull(Joiner.MINIMUM, minimum)
.putIfNotNull(Joiner.MAXIMUM, maximum)
@@ -757,6 +774,9 @@ public class Enrichers {
public AggregatorBuilder(AttributeSensor<S> aggregating) {
super(aggregating);
}
+ public AggregatorBuilder(AttributeSensor<S> keySensor, AttributeSensor<S> valueSensor) {
+ super(keySensor, valueSensor);
+ }
}
public static class PropagatorBuilder extends AbstractPropagatorBuilder<PropagatorBuilder> {
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/c58fe29e/core/src/main/java/org/apache/brooklyn/enricher/stock/Joiner.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/enricher/stock/Joiner.java b/core/src/main/java/org/apache/brooklyn/enricher/stock/Joiner.java
index 2e24676..02654bc 100644
--- a/core/src/main/java/org/apache/brooklyn/enricher/stock/Joiner.java
+++ b/core/src/main/java/org/apache/brooklyn/enricher/stock/Joiner.java
@@ -50,6 +50,10 @@ public class Joiner<T> extends AbstractEnricher implements SensorEventListener<T
public static ConfigKey<Sensor<?>> TARGET_SENSOR = ConfigKeys.newConfigKey(new TypeToken<Sensor<?>>() {}, "enricher.targetSensor");
@SetFromFlag("separator")
public static ConfigKey<String> SEPARATOR = ConfigKeys.newStringConfigKey("enricher.joiner.separator", "Separator string to insert between each argument", ",");
+ @SetFromFlag("keyValueSeparator")
+ public static ConfigKey<String> KEY_VALUE_SEPARATOR = ConfigKeys.newStringConfigKey("enricher.joiner.keyValueSeparator", "Separator string to insert between each key-value pair", "=");
+ @SetFromFlag("joinMapEntries")
+ public static ConfigKey<Boolean> JOIN_MAP_ENTRIES = ConfigKeys.newBooleanConfigKey("enricher.joiner.joinMapEntries", "Whether to add map entries as key-value pairs or just use the value, defaulting to false", false);
@SetFromFlag("quote")
public static ConfigKey<Boolean> QUOTE = ConfigKeys.newBooleanConfigKey("enricher.joiner.quote", "Whether to bash-escape each parameter and wrap in double-quotes, defaulting to true", true);
@SetFromFlag("minimum")
@@ -72,12 +76,12 @@ public class Joiner<T> extends AbstractEnricher implements SensorEventListener<T
this.producer = getConfig(PRODUCER) == null ? entity: getConfig(PRODUCER);
this.sourceSensor = (AttributeSensor<T>) getRequiredConfig(SOURCE_SENSOR);
this.targetSensor = (Sensor<String>) getRequiredConfig(TARGET_SENSOR);
-
+
subscriptions().subscribe(producer, sourceSensor, this);
-
- Object value = producer.getAttribute((AttributeSensor<?>)sourceSensor);
+
+ Object value = producer.getAttribute((AttributeSensor<?>) sourceSensor);
// TODO would be useful to have a convenience to "subscribeAndThenIfItIsAlreadySetRunItOnce"
- if (value!=null) {
+ if (value != null) {
onEvent(new BasicSensorEvent(sourceSensor, producer, value, -1));
}
}
@@ -92,36 +96,48 @@ public class Joiner<T> extends AbstractEnricher implements SensorEventListener<T
Object result = null;
if (v!=null) {
if (v instanceof Map) {
- v = ((Map<?,?>)v).values();
+ if (config().get(JOIN_MAP_ENTRIES)) {
+ v = ((Map<?,?>) v).entrySet();
+ } else {
+ v = ((Map<?,?>) v).values();
+ }
}
if (!(v instanceof Iterable)) {
LOG.warn("Enricher "+this+" received a non-iterable value "+v.getClass()+" "+v+"; refusing to join");
} else {
MutableList<Object> c1 = MutableList.of();
- Integer maximum = getConfig(MAXIMUM);
+ Integer maximum = config().get(MAXIMUM);
for (Object ci: (Iterable<?>)v) {
if (maximum!=null && maximum>=0) {
if (c1.size()>=maximum) break;
}
- c1.appendIfNotNull(Strings.toString(ci));
+ if (ci instanceof Map.Entry) {
+ String key = Strings.toString(((Map.Entry) ci).getKey());
+ Object value = ((Map.Entry) ci).getValue();
+ String keyValueSeparator = config().get(KEY_VALUE_SEPARATOR);
+ if (value != null) {
+ c1.append(String.format("%s%s%s", key, keyValueSeparator, Strings.toString(value)));
+ }
+ } else {
+ c1.appendIfNotNull(Strings.toString(ci));
+ }
}
- Integer minimum = getConfig(MINIMUM);
+ Integer minimum = config().get(MINIMUM);
if (minimum!=null && c1.size() < minimum) {
// use default null return value
} else {
- if (getConfig(QUOTE)) {
+ if (config().get(QUOTE)) {
MutableList<Object> c2 = MutableList.of();
for (Object ci: c1) {
c2.add(StringEscapes.BashStringEscapes.wrapBash((String)ci));
}
c1 = c2;
}
- result = Strings.join(c1, getConfig(SEPARATOR));
+ result = Strings.join(c1, config().get(SEPARATOR));
}
}
}
- if (LOG.isTraceEnabled())
- LOG.trace("Enricher "+this+" computed "+result+" from "+event);
+ LOG.trace("Enricher "+this+" computed "+result+" from "+event);
return result;
}
}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/c58fe29e/core/src/main/java/org/apache/brooklyn/enricher/stock/MapAggregator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/enricher/stock/MapAggregator.java b/core/src/main/java/org/apache/brooklyn/enricher/stock/MapAggregator.java
new file mode 100644
index 0000000..af1cf20
--- /dev/null
+++ b/core/src/main/java/org/apache/brooklyn/enricher/stock/MapAggregator.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.brooklyn.enricher.stock;
+
+import java.util.Collection;
+import java.util.Map;
+
+import com.google.common.base.Predicate;
+import com.google.common.base.Predicates;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Maps;
+import com.google.common.reflect.TypeToken;
+
+import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.api.sensor.Sensor;
+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.text.StringPredicates;
+
+/**
+ * Building on {@link AbstractMultipleSensorAggregator} for a pair of source sensors(on multiple children and/or members)
+ * that are used as key-value pairs in a generated Map.
+ */
+@SuppressWarnings("serial")
+public class MapAggregator<U> extends AbstractMultipleSensorAggregator<U> {
+
+ public static final ConfigKey<Sensor<?>> KEY_SENSOR = ConfigKeys.newConfigKey(new TypeToken<Sensor<?>>() {}, "enricher.keySensor");
+ public static final ConfigKey<Sensor<?>> VALUE_SENSOR = ConfigKeys.newConfigKey(new TypeToken<Sensor<?>>() {}, "enricher.valueSensor");
+
+ public static final ConfigKey<Boolean> EXCLUDE_BLANK = Aggregator.EXCLUDE_BLANK;
+
+ private Sensor<?> keySensor;
+ private Sensor<?> valueSensor;
+
+ public MapAggregator() { }
+
+ @Override
+ protected Predicate<?> getDefaultValueFilter() {
+ if (config().get(EXCLUDE_BLANK)) {
+ return StringPredicates.isNonBlank();
+ } else {
+ return Predicates.alwaysTrue();
+ }
+ }
+
+ @Override
+ protected Object compute() {
+ Map<Entity, Object> ks = MutableMap.copyOf(Maps.filterValues(getValues(keySensor), valueFilter));
+ Map<Entity, Object> vs = MutableMap.copyOf(Maps.filterValues(getValues(valueSensor), valueFilter));
+ MutableMap<Object, Object> result = MutableMap.of();
+ for (Entity entity : ks.keySet()) {
+ if (vs.containsKey(entity)) {
+ result.put(ks.get(entity), vs.get(entity));
+ }
+ }
+ return result;
+ }
+
+ @Override
+ protected Collection<Sensor<?>> getSourceSensors() {
+ keySensor = config().get(KEY_SENSOR);
+ valueSensor = config().get(VALUE_SENSOR);
+ return ImmutableList.<Sensor<?>>of(keySensor, valueSensor);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/c58fe29e/core/src/test/java/org/apache/brooklyn/enricher/stock/EnrichersTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/brooklyn/enricher/stock/EnrichersTest.java b/core/src/test/java/org/apache/brooklyn/enricher/stock/EnrichersTest.java
index 6e5cc16..e20664b 100644
--- a/core/src/test/java/org/apache/brooklyn/enricher/stock/EnrichersTest.java
+++ b/core/src/test/java/org/apache/brooklyn/enricher/stock/EnrichersTest.java
@@ -393,6 +393,29 @@ public class EnrichersTest extends BrooklynAppUnitTestSupport {
entity.sensors().set(NUM1, 987654);
EntityAsserts.assertAttributeEqualsContinually(group, LONG1, Long.valueOf(123));
}
+
+ @Test
+ public void testAggregatingMap() {
+ group.addMember(entity);
+ group.enrichers().add(Enrichers.builder()
+ .aggregating(STR1, STR2)
+ .publishing(MAP1)
+ .fromMembers()
+ .excludingBlank()
+ .build());
+
+ EntityAsserts.assertAttributeEqualsEventually(group, MAP1, ImmutableMap.<String, String>of());
+
+ entity.sensors().set(STR1, "a");
+ EntityAsserts.assertAttributeEqualsEventually(group, MAP1, ImmutableMap.<String, String>of());
+
+ entity.sensors().set(STR2, "b");
+ EntityAsserts.assertAttributeEqualsEventually(group, MAP1, ImmutableMap.<String, String>of("a", "b"));
+
+ entity.sensors().set(STR2, "c");
+ EntityAsserts.assertAttributeEqualsEventually(group, MAP1, ImmutableMap.<String, String>of("a", "c"));
+ }
+
@Test
public void testUpdatingMap1() {
entity.enrichers().add(Enrichers.builder()
@@ -429,6 +452,7 @@ public class EnrichersTest extends BrooklynAppUnitTestSupport {
}
private static AttributeSensor<Object> LIST_SENSOR = Sensors.newSensor(Object.class, "sensor.list");
+ private static AttributeSensor<Object> MAP_SENSOR = Sensors.newSensor(Object.class, "sensor.map");
@Test
public void testJoinerDefault() {
@@ -490,5 +514,21 @@ public class EnrichersTest extends BrooklynAppUnitTestSupport {
EntityAsserts.assertAttributeEqualsEventually(entity, TestEntity.NAME, "a,b,c,d");
}
+ @Test
+ public void testJoinerMap() {
+ entity.enrichers().add(Enrichers.builder()
+ .joining(MAP_SENSOR)
+ .keyValueSeparator("=")
+ .joinMapEntries(true)
+ .publishing(TestEntity.NAME)
+ .build());
+ // check quotes
+ entity.sensors().set(MAP_SENSOR, MutableMap.<String, String>of("a", "\"v", "b", "w x y"));
+ EntityAsserts.assertAttributeEqualsEventually(entity, TestEntity.NAME, "\"a=\\\"v\",\"b=w x y\"");
+
+ // empty map causes ""
+ entity.sensors().set(MAP_SENSOR, MutableMap.<String, String>of());
+ EntityAsserts.assertAttributeEqualsEventually(entity, TestEntity.NAME, "");
+ }
}