cleanup, and allow TaskFactory to be supplied as a config and other ValueResolver input

the TF will create a task which will then be used for evaluation.
much cleaner semantics than setting tasks as values:

tasks evaluate once and remember their result, whereas task factory spawns a new task each time.
furthermore, the former cannot be interrupted without making the value _never_ resolvable (which was the case prior to the previous commit)
so it is left running if immediate eval is done, whereas the latter can be safely cancelled.

Author: Alex Heneveld

camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java

@@ -203,6 +203,15 @@ public Maybe<Entity> getImmediately() {
             }
         }
 
+        @Override
+        public Entity get() {
+            try {
+                return call();
+            } catch (Exception e) {
+                throw Exceptions.propagate(e);
+            }
+        }
+        
         @Override
         public Entity call() throws Exception {
             return callImpl(false).get();
@@ -304,10 +313,11 @@ protected Maybe<Entity> callImpl(boolean immediate) throws Exception {
                 return Maybe.of(result.get());
             }
 
-            // TODO may want to block and repeat on new entities joining?
-            throw new NoSuchElementException("No entity matching id " + desiredComponentId+
+            // could be nice if DSL has an extra .block() method to allow it to wait for a matching entity.
+            // previously we threw if nothing existed; now we return an absent with a detailed error
+            return Maybe.absent(new NoSuchElementException("No entity matching id " + desiredComponentId+
                 (scope==Scope.GLOBAL ? "" : ", in scope "+scope+" wrt "+entity+
-                (scopeComponent!=null ? " ("+scopeComponent+" from "+entity()+")" : "")));
+                (scopeComponent!=null ? " ("+scopeComponent+" from "+entity()+")" : ""))));
         }
 
         private ExecutionContext getExecutionContext() {

camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslTest.java

@@ -48,6 +48,7 @@
 import org.apache.brooklyn.util.guava.Maybe;
 import org.apache.brooklyn.util.text.Identifiers;
 import org.apache.brooklyn.util.time.Duration;
+import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
@@ -296,14 +297,13 @@ public void testUrlEncode() throws Exception {
     @Test
     public void testEntityNotFound() throws Exception {
         BrooklynDslDeferredSupplier<?> dsl = BrooklynDslCommon.entity("myIdDoesNotExist");
+        Maybe<?> actualValue = execDslImmediately(dsl, Entity.class, app, true);
+        Assert.assertTrue(actualValue.isAbsent());
         try {
-            Maybe<?> actualValue = execDslImmediately(dsl, Entity.class, app, true);
+            actualValue.get();
             Asserts.shouldHaveFailedPreviously("actual="+actualValue);
         } catch (Exception e) {
-            NoSuchElementException nsee = Exceptions.getFirstThrowableOfType(e, NoSuchElementException.class);
-            if (nsee == null) {
-                throw e;
-            }
+            Asserts.expectedFailureOfType(e, NoSuchElementException.class);
         }
     }
 
@@ -365,6 +365,7 @@ public DslTestWorker satisfiedAsynchronously(boolean val) {
             return this;
         }
 
+        @SuppressWarnings("unused")  // kept in case useful for additional tests
         public DslTestWorker wrapInTaskForImmediately(boolean val) {
             wrapInTaskForImmediately = val;
             return this;

core/src/main/java/org/apache/brooklyn/core/config/internal/AbstractConfigMapImpl.java

@@ -30,6 +30,7 @@
 import javax.annotation.Nullable;
 
 import org.apache.brooklyn.api.mgmt.ExecutionContext;
+import org.apache.brooklyn.api.mgmt.TaskFactory;
 import org.apache.brooklyn.api.objs.BrooklynObject;
 import org.apache.brooklyn.config.ConfigInheritance;
 import org.apache.brooklyn.config.ConfigInheritances;
@@ -231,7 +232,7 @@ public Maybe<Object> getConfigRaw(ConfigKey<?> key, boolean includeInherited) {
     }
 
     protected Object coerceConfigVal(ConfigKey<?> key, Object v) {
-        if ((v instanceof Future) || (v instanceof DeferredSupplier)) {
+        if ((v instanceof Future) || (v instanceof DeferredSupplier) || (v instanceof TaskFactory)) {
             // no coercion for these (coerce on exit)
             return v;
         } else if (key instanceof StructuredConfigKey) {

core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java

@@ -46,7 +46,6 @@
 import org.slf4j.LoggerFactory;
 
 import com.google.common.base.Function;
-import com.google.common.base.Supplier;
 import com.google.common.collect.Iterables;
 
 /**
@@ -99,7 +98,8 @@ public ExecutionManager getExecutionManager() {
     @Override
     public Set<Task<?>> getTasks() { return executionManager.getTasksWithAllTags(tags); }
 
-    /** performs execution without spawning a new task thread, though it does temporarily set a fake task for the purpose of getting context */
+    /** performs execution without spawning a new task thread, though it does temporarily set a fake task for the purpose of getting context;
+     * currently supports suppliers or callables  */
     @SuppressWarnings("unchecked")
     @Override
     public <T> Maybe<T> getImmediately(Object callableOrSupplier) {
@@ -110,17 +110,15 @@ public <T> Maybe<T> getImmediately(Object callableOrSupplier) {
 
         Task<?> previousTask = BasicExecutionManager.getPerThreadCurrentTask().get();
         if (previousTask!=null) fakeTaskForContext.setSubmittedByTask(previousTask);
+        fakeTaskForContext.cancel();
         try {
             BasicExecutionManager.getPerThreadCurrentTask().set(fakeTaskForContext);
 
-            if ((callableOrSupplier instanceof Supplier) && !(callableOrSupplier instanceof ImmediateSupplier)) {
-                callableOrSupplier = new InterruptingImmediateSupplier<>((Supplier<Object>)callableOrSupplier);
+            if (!(callableOrSupplier instanceof ImmediateSupplier)) {
+                callableOrSupplier = InterruptingImmediateSupplier.of(callableOrSupplier);
             }
-            if (callableOrSupplier instanceof ImmediateSupplier) {
-                return ((ImmediateSupplier<T>)callableOrSupplier).getImmediately();
-            }
-            // TODO could add more types here
-            throw new IllegalArgumentException("Type "+callableOrSupplier.getClass()+" not supported for getImmediately (instance "+callableOrSupplier+")");
+            return ((ImmediateSupplier<T>)callableOrSupplier).getImmediately();
+ 
         } finally {
             BasicExecutionManager.getPerThreadCurrentTask().set(previousTask);
         }

core/src/main/java/org/apache/brooklyn/util/core/task/ImmediateSupplier.java

@@ -20,11 +20,13 @@
 
 import org.apache.brooklyn.util.guava.Maybe;
 
+import com.google.common.base.Supplier;
+
 /**
- * A class that supplies objects of a single type, without blocking for any significant length
- * of time.
+ * A {@link Supplier} that has an extra method capable of supplying a value immediately or an absent if definitely not available,
+ * or throwing an {@link ImmediateUnsupportedException} if it cannot determine whether a value is immediately available.
  */
-public interface ImmediateSupplier<T> {
+public interface ImmediateSupplier<T> extends Supplier<T> {
 
     /**
      * Indicates that a supplier does not support immediate evaluation,

core/src/main/java/org/apache/brooklyn/util/core/task/InterruptingImmediateSupplier.java

@@ -18,13 +18,14 @@
  */
 package org.apache.brooklyn.util.core.task;
 
-import java.util.NoSuchElementException;
+import java.util.concurrent.Callable;
 import java.util.concurrent.Semaphore;
 
 import org.apache.brooklyn.util.exceptions.Exceptions;
 import org.apache.brooklyn.util.exceptions.RuntimeInterruptedException;
 import org.apache.brooklyn.util.guava.Maybe;
 
+import com.google.common.annotations.Beta;
 import com.google.common.base.Supplier;
 
 /**
@@ -40,6 +41,7 @@
  * will throw if the thread is interrupted.  Typically there are workarounds, for instance:
  * <code>if (semaphore.tryAcquire()) semaphore.acquire();</code>. 
  */
+@Beta
 public class InterruptingImmediateSupplier<T> implements ImmediateSupplier<T>, DeferredSupplier<T> {
 
     final Supplier<T> nestedSupplier;
@@ -69,6 +71,33 @@ public Maybe<T> getImmediately() {
     public T get() {
         return nestedSupplier.get();
     }
-    
+
+    @SuppressWarnings("unchecked")
+    public static <T> InterruptingImmediateSupplier<T> of(final Object o) {
+        if (o instanceof Supplier) {
+            return new InterruptingImmediateSupplier<T>((Supplier<T>)o);
+        } else if (o instanceof Callable) {
+            return new InterruptingImmediateSupplier<T>(new Supplier<T>() {
+                @Override
+                public T get() {
+                    try {
+                        return ((Callable<T>)o).call();
+                    } catch (Exception e) {
+                        throw Exceptions.propagate(e);
+                    }
+                }
+            });
+        } else if (o instanceof Runnable) {
+            return new InterruptingImmediateSupplier<T>(new Supplier<T>() {
+                @Override
+                public T get() {
+                    ((Runnable)o).run();
+                    return null;
+                }
+            });
+        } else {
+            throw new UnsupportedOperationException("Type "+o.getClass()+" not supported as InterruptingImmediateSupplier (instance "+o+")");
+        }
+    }
 
 }

core/src/main/java/org/apache/brooklyn/util/core/task/TaskTags.java

@@ -62,6 +62,7 @@ public static boolean isInessential(Task<?> task) {
     }
 
     public static boolean hasTag(Task<?> task, Object tag) {
+        if (task==null) return false;
         return task.getTags().contains(tag);
     }
 

core/src/main/java/org/apache/brooklyn/util/core/task/ValueResolver.java

@@ -29,6 +29,7 @@
 import org.apache.brooklyn.api.mgmt.ExecutionContext;
 import org.apache.brooklyn.api.mgmt.Task;
 import org.apache.brooklyn.api.mgmt.TaskAdaptable;
+import org.apache.brooklyn.api.mgmt.TaskFactory;
 import org.apache.brooklyn.core.entity.EntityInternal;
 import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
 import org.apache.brooklyn.util.core.flags.TypeCoercions;
@@ -322,6 +323,10 @@ public Maybe<T> getMaybe() {
         return result;
     }
 
+    protected boolean isEvaluatingImmediately() {
+        return immediately || BrooklynTaskTags.hasTag(Tasks.current(), BrooklynTaskTags.IMMEDIATE_TASK_TAG);
+    }
+    
     @SuppressWarnings({ "unchecked", "rawtypes" })
     protected Maybe<T> getMaybeInternal() {
         if (started.getAndSet(true))
@@ -352,11 +357,11 @@ protected Maybe<T> getMaybeInternal() {
 
         //if the expected type is a closure or map and that's what we have, we're done (or if it's null);
         //but not allowed to return a future or DeferredSupplier as the resolved value
-        if (v==null || (!forceDeep && type.isInstance(v) && !Future.class.isInstance(v) && !DeferredSupplier.class.isInstance(v)))
+        if (v==null || (!forceDeep && type.isInstance(v) && !Future.class.isInstance(v) && !DeferredSupplier.class.isInstance(v) && !TaskFactory.class.isInstance(v)))
             return Maybe.of((T) v);
 
         try {
-            if (immediately && v instanceof ImmediateSupplier) {
+            if (isEvaluatingImmediately() && v instanceof ImmediateSupplier) {
                 final ImmediateSupplier<Object> supplier = (ImmediateSupplier<Object>) v;
                 try {
                     Maybe<Object> result = exec.getImmediately(supplier);
@@ -366,12 +371,23 @@ protected Maybe<T> getMaybeInternal() {
                             ? recursive
                                 ? new ValueResolver(result.get(), type, this).getMaybe()
                                 : result
-                            : Maybe.<T>absent();
+                            : result;
                 } catch (ImmediateSupplier.ImmediateUnsupportedException e) {
                     log.debug("Unable to resolve-immediately for "+description+" ("+v+"); falling back to executing with timeout", e);
                 }
             }
 
+            // TODO if evaluating immediately should use a new ExecutionContext.submitImmediate(...)
+            // and sets a timeout but which wraps a task but does not spawn a new thread
+            
+            if ((v instanceof TaskFactory<?>) && !(v instanceof DeferredSupplier)) {
+                v = ((TaskFactory<?>)v).newTask();
+                BrooklynTaskTags.setTransient(((TaskAdaptable<?>)v).asTask());
+                if (isEvaluatingImmediately()) {
+                    BrooklynTaskTags.addTagDynamically( ((TaskAdaptable<?>)v).asTask(), BrooklynTaskTags.IMMEDIATE_TASK_TAG );
+                }
+            }
+            
             //if it's a task or a future, we wait for the task to complete
             if (v instanceof TaskAdaptable<?>) {
                 //if it's a task, we make sure it is submitted
@@ -382,7 +398,7 @@ protected Maybe<T> getMaybeInternal() {
                     }
                     if (!task.getTags().contains(BrooklynTaskTags.TRANSIENT_TASK_TAG)) {
                         // mark this non-transient, because this value is usually something set e.g. in config
-                        // (ideally we'd discourage this in favour of task factories which can be transiently interrupted)
+                        // (should discourage this in favour of task factories which can be transiently interrupted?)
                         BrooklynTaskTags.addTagDynamically(task, BrooklynTaskTags.NON_TRANSIENT_TASK_TAG);
                     }
                     exec.submit(task);

core/src/test/java/org/apache/brooklyn/core/entity/EntityConfigTest.java

@@ -313,12 +313,12 @@ private DeferredSupplier<String> deferredSupplier() {
             return new DeferredSupplier<String>() {
                 @Override public String get() {
                     try {
-                        log.info("acquiring");
+                        log.trace("acquiring");
                         if (!latch.tryAcquire()) latch.acquire();
                         latch.release();
-                        log.info("acquired and released");
+                        log.trace("acquired and released");
                     } catch (InterruptedException e) {
-                        log.info("interrupted");
+                        log.trace("interrupted");
                         throw Exceptions.propagate(e);
                     }
                     return "myval";
@@ -333,21 +333,21 @@ protected void runGetConfigNonBlockingInKey() throws Exception {
             TestEntity entity = (TestEntity) mgmt.getEntityManager().createEntity(EntitySpec.create(TestEntity.class)
                     .configure((ConfigKey<Object>)(ConfigKey<?>)TestEntity.CONF_NAME, blockingVal));
 
-            log.info("get non-blocking");
+            log.trace("get non-blocking");
             // Will initially return absent, because task is not done
             assertTrue(entity.config().getNonBlocking(TestEntity.CONF_NAME).isAbsent());
-            log.info("got absent");
+            log.trace("got absent");
 
             latch.release();
 
             // Can now finish task, so will return expectedVal
-            log.info("get blocking");
+            log.trace("get blocking");
             assertEquals(entity.config().get(TestEntity.CONF_NAME), expectedVal);
-            log.info("got blocking");
+            log.trace("got blocking");
             assertEquals(entity.config().getNonBlocking(TestEntity.CONF_NAME).get(), expectedVal);
 
             latch.acquire();
-            log.info("finished");
+            log.trace("finished");
         }
 
         protected void runGetConfigNonBlockingInMap() throws Exception {
@@ -526,7 +526,7 @@ public String call() {
         assertEquals(getConfigFuture.get(TIMEOUT_MS, TimeUnit.MILLISECONDS), "abc");
     }
 
-    @Test
+    @Test(groups="Integration")  // takes 0.5s
     public void testGetConfigWithExecutedTaskWaitsForResult() throws Exception {
         LatchingCallable<String> work = new LatchingCallable<String>("abc");
         Task<String> task = executionManager.submit(work);
@@ -548,7 +548,7 @@ public String call() {
         assertEquals(work.callCount.get(), 1);
     }
 
-    @Test
+    @Test(groups="Integration")  // takes 0.5s
     public void testGetConfigWithUnexecutedTaskIsExecutedAndWaitsForResult() throws Exception {
         LatchingCallable<String> work = new LatchingCallable<String>("abc");
         Task<String> task = new BasicTask<String>(work);