You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@brooklyn.apache.org by al...@apache.org on 2017/11/06 23:20:25 UTC
[1/3] brooklyn-server git commit: failing test illustrating problem
Repository: brooklyn-server
Updated Branches:
refs/heads/master 2f57d79eb -> b1126924f
failing test illustrating problem
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/e797cbe1
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/e797cbe1
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/e797cbe1
Branch: refs/heads/master
Commit: e797cbe179e0483599474a66141ed30d5f8c607a
Parents: 48f0900
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Thu Nov 2 15:35:58 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Thu Nov 2 15:36:15 2017 +0000
----------------------------------------------------------------------
.../mgmt/internal/LocalSubscriptionManagerTest.java | 16 ++++++++++++++++
1 file changed, 16 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/e797cbe1/core/src/test/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManagerTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManagerTest.java b/core/src/test/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManagerTest.java
index 5c04978..09cd40e 100644
--- a/core/src/test/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManagerTest.java
+++ b/core/src/test/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManagerTest.java
@@ -41,6 +41,7 @@ import org.apache.brooklyn.core.test.entity.TestEntity;
import org.apache.brooklyn.entity.group.BasicGroup;
import org.apache.brooklyn.test.Asserts;
import org.apache.brooklyn.util.collections.MutableMap;
+import org.apache.brooklyn.util.core.task.BasicExecutionContext;
import org.apache.brooklyn.util.time.Duration;
import org.testng.Assert;
import org.testng.annotations.BeforeMethod;
@@ -259,4 +260,19 @@ public class LocalSubscriptionManagerTest extends BrooklynAppUnitTestSupport {
Assert.assertEquals(listener.getEvents().get(9).getValue(), entity.sensors().get(TestEntity.SEQUENCE));
}
+ @Test
+ public void testSubscriptionHasSubscribersExecutionContext() throws Exception {
+ final AtomicReference<BasicExecutionContext> result = new AtomicReference<>();
+ final CountDownLatch latch = new CountDownLatch(1);
+ app.subscriptions().subscribe(entity, TestEntity.SEQUENCE, new SensorEventListener<Object>() {
+ @Override public void onEvent(SensorEvent<Object> event) {
+ result.set(BasicExecutionContext.getCurrentExecutionContext());
+ latch.countDown();
+ }});
+ entity.setSequenceValue(1234);
+ if (!latch.await(TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
+ fail("Timeout waiting for Event on TestEntity listener");
+ }
+ Assert.assertEquals(result.get(), app.getExecutionContext());
+ }
}
[2/3] brooklyn-server git commit: share code for inferring execution
context, and ensure EC is set in subscriptions
Posted by al...@apache.org.
share code for inferring execution context, and ensure EC is set in subscriptions
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/95c5b1e7
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/95c5b1e7
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/95c5b1e7
Branch: refs/heads/master
Commit: 95c5b1e78fba24e37216e99c209568a5e3cf2fec
Parents: e797cbe
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Thu Nov 2 15:20:17 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Thu Nov 2 15:36:16 2017 +0000
----------------------------------------------------------------------
.../spi/dsl/BrooklynDslDeferredSupplier.java | 5 +-
.../brooklyn/spi/dsl/methods/DslComponent.java | 6 +--
.../brooklyn/core/mgmt/BrooklynTaskTags.java | 49 ++++++++++++++++++++
.../mgmt/internal/LocalSubscriptionManager.java | 7 +++
.../util/core/task/BasicExecutionContext.java | 12 ++++-
.../customize/TemplateOptionsOption.java | 10 +---
6 files changed, 70 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/95c5b1e7/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java
----------------------------------------------------------------------
diff --git a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java
index 835f5d4..3b31637 100644
--- a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java
+++ b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java
@@ -100,10 +100,7 @@ public abstract class BrooklynDslDeferredSupplier<T> implements DeferredSupplier
if (log.isDebugEnabled())
log.debug("Queuing task to resolve "+dsl+", called by "+Tasks.current());
- EntityInternal entity = (EntityInternal) BrooklynTaskTags.getTargetOrContextEntity(Tasks.current());
- ExecutionContext exec =
- (entity != null) ? entity.getExecutionContext()
- : BasicExecutionContext.getCurrentExecutionContext();
+ ExecutionContext exec = BrooklynTaskTags.getCurrentExecutionContext();
if (exec == null) {
throw new IllegalStateException("No execution context available to resolve " + dsl);
}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/95c5b1e7/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java
----------------------------------------------------------------------
diff --git a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java
index 9ecad76..9dc5ed1 100644
--- a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java
+++ b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java
@@ -46,7 +46,6 @@ import org.apache.brooklyn.core.sensor.DependentConfiguration;
import org.apache.brooklyn.core.sensor.Sensors;
import org.apache.brooklyn.util.JavaGroovyEquivalents;
import org.apache.brooklyn.util.core.flags.TypeCoercions;
-import org.apache.brooklyn.util.core.task.BasicExecutionContext;
import org.apache.brooklyn.util.core.task.DeferredSupplier;
import org.apache.brooklyn.util.core.task.ImmediateSupplier;
import org.apache.brooklyn.util.core.task.TaskBuilder;
@@ -339,10 +338,7 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
}
static ExecutionContext findExecutionContext(Object callerContext) {
- EntityInternal contextEntity = (EntityInternal) BrooklynTaskTags.getTargetOrContextEntity(Tasks.current());
- ExecutionContext execContext =
- (contextEntity != null) ? contextEntity.getExecutionContext()
- : BasicExecutionContext.getCurrentExecutionContext();
+ ExecutionContext execContext = BrooklynTaskTags.getCurrentExecutionContext();
if (execContext == null) {
throw new IllegalStateException("No execution context available to resolve " + callerContext);
}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/95c5b1e7/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java b/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java
index f8a9d0b..b9dfbc1 100644
--- a/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java
+++ b/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java
@@ -30,13 +30,17 @@ import javax.annotation.Nullable;
import org.apache.brooklyn.api.effector.Effector;
import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.api.mgmt.ExecutionContext;
import org.apache.brooklyn.api.mgmt.ExecutionManager;
import org.apache.brooklyn.api.mgmt.ManagementContext;
import org.apache.brooklyn.api.mgmt.Task;
import org.apache.brooklyn.api.mgmt.entitlement.EntitlementContext;
import org.apache.brooklyn.api.objs.EntityAdjunct;
+import org.apache.brooklyn.core.entity.EntityInternal;
import org.apache.brooklyn.core.mgmt.internal.AbstractManagementContext;
+import org.apache.brooklyn.core.objs.AbstractEntityAdjunct;
import org.apache.brooklyn.util.core.config.ConfigBag;
+import org.apache.brooklyn.util.core.task.BasicExecutionContext;
import org.apache.brooklyn.util.core.task.DynamicTasks;
import org.apache.brooklyn.util.core.task.TaskTags;
import org.apache.brooklyn.util.core.task.Tasks;
@@ -183,11 +187,25 @@ public class BrooklynTaskTags extends TaskTags {
return getWrappedEntityTagOfType( getTagsFast(t), wrappingType);
}
public static WrappedEntity getWrappedEntityTagOfType(Collection<?> tags, String wrappingType) {
+ if (tags==null) return null;
for (Object x: tags)
if ((x instanceof WrappedEntity) && ((WrappedEntity)x).wrappingType.equals(wrappingType))
return (WrappedEntity)x;
return null;
}
+
+ @SuppressWarnings("unchecked")
+ public static <T> WrappedObject<T> getWrappedObjectTagOfType(Collection<?> tags, String wrappingType, Class<T> type) {
+ if (tags==null) return null;
+ for (Object x: tags)
+ if ((x instanceof WrappedObject) && ((WrappedObject<?>)x).wrappingType.equals(wrappingType) && type.isInstance( ((WrappedObject<?>)x).object ))
+ return (WrappedObject<T>)x;
+ return null;
+ }
+ public static <T> T getUnwrappedObjectTagOfType(Collection<?> tags, String wrappingType, Class<T> type) {
+ WrappedObject<T> result = getWrappedObjectTagOfType(tags, wrappingType, type);
+ return result!=null ? result.unwrap() : null;
+ }
public static Entity getWrappedEntityOfType(Task<?> t, String wrappingType) {
WrappedEntity wrapper = getWrappedEntityTagOfType(t, wrappingType);
@@ -540,5 +558,36 @@ public class BrooklynTaskTags extends TaskTags {
tag.entitlementContext = context;
return tag;
}
+
+ public static ExecutionContext getExecutionContext(Collection<?> tags) {
+ EntityAdjunct ea = getUnwrappedObjectTagOfType(tags, CONTEXT_ADJUNCT, EntityAdjunct.class);
+ if (ea instanceof AbstractEntityAdjunct) {
+ return ((AbstractEntityAdjunct)ea).getExecutionContext();
+ }
+ Entity e = getWrappedEntityOfType(tags, CONTEXT_ENTITY);
+ if (e==null) e= getWrappedEntityOfType(tags, TARGET_ENTITY);
+ if (e instanceof EntityInternal) {
+ return ((EntityInternal)e).getExecutionContext();
+ }
+
+ return null;
+ }
+ public static ExecutionContext getExecutionContext(Task<?> t) {
+ return getExecutionContext(getTagsFast(t));
+ }
+ /**
+ * This should always be identical to {@link BasicExecutionContext#getCurrentExecutionContext()} as
+ * there are two ways to derive the {@link ExecutionContext}: one is the {@link ThreadLocal} accessed from the above method
+ * and set by {@link BasicExecutionContext} submissions; the other is by looking at {@link #getTargetOrContextEntity(Task)}
+ * and {@link #CONTEXT_ADJUNCT} tags on the current task. As {@link BasicExecutionContext} also sets those tags,
+ * the only time the two will be different is if {@link ExecutionManager#submit(org.apache.brooklyn.api.mgmt.TaskAdaptable)}
+ * is used supplying one or more of these context tags, in which case only this method will determine the execution context.
+ */
+ public static ExecutionContext getCurrentExecutionContext() {
+ Task<?> t = Tasks.current();
+ ExecutionContext result = t!=null ? getExecutionContext(t) : null;
+ if (result==null) result = BasicExecutionContext.getCurrentExecutionContext();
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/95c5b1e7/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManager.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManager.java b/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManager.java
index 623fb26..a5fed0e 100644
--- a/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManager.java
+++ b/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManager.java
@@ -34,6 +34,7 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.api.mgmt.ExecutionContext;
import org.apache.brooklyn.api.mgmt.ExecutionManager;
import org.apache.brooklyn.api.mgmt.SubscriptionHandle;
import org.apache.brooklyn.api.mgmt.SubscriptionManager;
@@ -48,6 +49,7 @@ import org.apache.brooklyn.core.sensor.AttributeMap;
import org.apache.brooklyn.core.sensor.BasicSensorEvent;
import org.apache.brooklyn.util.collections.MutableList;
import org.apache.brooklyn.util.collections.MutableMap;
+import org.apache.brooklyn.util.core.task.BasicExecutionContext;
import org.apache.brooklyn.util.core.task.BasicExecutionManager;
import org.apache.brooklyn.util.core.task.SingleThreadedScheduler;
import org.apache.brooklyn.util.text.Identifiers;
@@ -311,6 +313,7 @@ public class LocalSubscriptionManager extends AbstractSubscriptionManager {
boolean isEntityStarting = s.subscriber instanceof Entity && isInitialPublicationOfOldValueInCorrectScheduledThread;
// will have entity (and adjunct) execution context from tags, so can skip getting exec context
+ final ExecutionContext ec = BrooklynTaskTags.getExecutionContext(tags);
Runnable deliverer = new Runnable() {
@Override
public String toString() {
@@ -322,7 +325,9 @@ public class LocalSubscriptionManager extends AbstractSubscriptionManager {
}
@Override
public void run() {
+ BasicExecutionContext oldEC = ec instanceof BasicExecutionContext ? BasicExecutionContext.setPerThreadExecutionContext((BasicExecutionContext)ec) : null;
try {
+
if (isEntityStarting) {
/* don't let sub deliveries start until this is completed;
* this is a pragmatic way to ensure the publish events
@@ -349,6 +354,8 @@ public class LocalSubscriptionManager extends AbstractSubscriptionManager {
} else {
LOG.warn("Error processing subscriptions to "+this+": "+t, t);
}
+ } finally {
+ BasicExecutionContext.setPerThreadExecutionContext(oldEC);
}
}};
if (!isInitialPublicationOfOldValueInCorrectScheduledThread) {
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/95c5b1e7/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
index f81ad35..8b37e2f 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
@@ -56,6 +56,7 @@ import org.apache.brooklyn.util.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.annotations.Beta;
import com.google.common.base.Function;
import com.google.common.base.Supplier;
import com.google.common.collect.Iterables;
@@ -69,7 +70,6 @@ public class BasicExecutionContext extends AbstractExecutionContext {
private static final Logger log = LoggerFactory.getLogger(BasicExecutionContext.class);
static final ThreadLocal<BasicExecutionContext> perThreadExecutionContext = new ThreadLocal<BasicExecutionContext>();
-
public static BasicExecutionContext getCurrentExecutionContext() { return perThreadExecutionContext.get(); }
final ExecutionManager executionManager;
@@ -448,6 +448,16 @@ public class BasicExecutionContext extends AbstractExecutionContext {
}
private void registerPerThreadExecutionContext() { perThreadExecutionContext.set(this); }
+ /** For use if external code wants to subsequently use an {@link ExecutionContext} but cannot submit via one.
+ * Caller should store the result and reset it back afterwards, in case a task may be running
+ * in the same thread as a synchronous submitter. */
+ // only LocalSubscriptionManager needs to do that; and it could be refactored to take the execution context rather than tags.
+ @Beta
+ public static BasicExecutionContext setPerThreadExecutionContext(BasicExecutionContext ec) {
+ BasicExecutionContext old = perThreadExecutionContext.get();
+ perThreadExecutionContext.set(ec);
+ return old;
+ }
private void clearPerThreadExecutionContext() { perThreadExecutionContext.remove(); }
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/95c5b1e7/locations/jclouds/src/main/java/org/apache/brooklyn/location/jclouds/templates/customize/TemplateOptionsOption.java
----------------------------------------------------------------------
diff --git a/locations/jclouds/src/main/java/org/apache/brooklyn/location/jclouds/templates/customize/TemplateOptionsOption.java b/locations/jclouds/src/main/java/org/apache/brooklyn/location/jclouds/templates/customize/TemplateOptionsOption.java
index c9b0e81..8174c22 100644
--- a/locations/jclouds/src/main/java/org/apache/brooklyn/location/jclouds/templates/customize/TemplateOptionsOption.java
+++ b/locations/jclouds/src/main/java/org/apache/brooklyn/location/jclouds/templates/customize/TemplateOptionsOption.java
@@ -23,14 +23,9 @@ import java.util.Map;
import java.util.concurrent.ExecutionException;
import org.apache.brooklyn.api.mgmt.ExecutionContext;
-import org.apache.brooklyn.api.mgmt.ManagementContext;
-import org.apache.brooklyn.api.mgmt.Task;
-import org.apache.brooklyn.core.entity.EntityInternal;
import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
-import org.apache.brooklyn.core.mgmt.internal.LocalManagementContext;
import org.apache.brooklyn.util.core.config.ConfigBag;
import org.apache.brooklyn.util.core.flags.MethodCoercions;
-import org.apache.brooklyn.util.core.task.BasicExecutionContext;
import org.apache.brooklyn.util.core.task.Tasks;
import org.apache.brooklyn.util.exceptions.Exceptions;
import org.apache.brooklyn.util.guava.Maybe;
@@ -53,10 +48,7 @@ public class TemplateOptionsOption implements TemplateOptionCustomizer {
if (optionValue != null) {
try {
- final EntityInternal entity = (EntityInternal) BrooklynTaskTags.getTargetOrContextEntity(Tasks.current());
- final ExecutionContext exec = (null != entity
- ? entity.getExecutionContext()
- : BasicExecutionContext.getCurrentExecutionContext());
+ final ExecutionContext exec = BrooklynTaskTags.getCurrentExecutionContext();
if (exec != null) {
optionValue = Tasks.resolveDeepValue(optionValue, Object.class, exec);
}
[3/3] brooklyn-server git commit: This closes #875
Posted by al...@apache.org.
This closes #875
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/b1126924
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/b1126924
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/b1126924
Branch: refs/heads/master
Commit: b1126924f42ba50721897f738ed1b1cbb4ef5774
Parents: 2f57d79 95c5b1e
Author: Aled Sage <al...@gmail.com>
Authored: Mon Nov 6 23:20:12 2017 +0000
Committer: Aled Sage <al...@gmail.com>
Committed: Mon Nov 6 23:20:12 2017 +0000
----------------------------------------------------------------------
.../spi/dsl/BrooklynDslDeferredSupplier.java | 5 +-
.../brooklyn/spi/dsl/methods/DslComponent.java | 6 +--
.../brooklyn/core/mgmt/BrooklynTaskTags.java | 49 ++++++++++++++++++++
.../mgmt/internal/LocalSubscriptionManager.java | 7 +++
.../util/core/task/BasicExecutionContext.java | 12 ++++-
.../internal/LocalSubscriptionManagerTest.java | 16 +++++++
.../customize/TemplateOptionsOption.java | 10 +---
7 files changed, 86 insertions(+), 19 deletions(-)
----------------------------------------------------------------------