[WIP] Add DSL support for calling effectors in YAML

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

@@ -70,6 +70,7 @@
 import com.google.common.base.Function;
 import com.google.common.base.Objects;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
@@ -233,6 +234,18 @@ public static BrooklynDslDeferredSupplier<?> entityId() {
         return new DslComponent(Scope.THIS, "").entityId();
     }
 
+    public static BrooklynDslDeferredSupplier<?> effector(String effectorName, Map<String, ?> args) {
+        return new DslComponent(Scope.THIS, "").effector(effectorName, args);
+    }
+
+    public static BrooklynDslDeferredSupplier<?> effector(String effectorName) {
+        return new DslComponent(Scope.THIS, "").effector(effectorName, ImmutableMap.<String, Object>of());
+    }
+
+    public static BrooklynDslDeferredSupplier<?> effector(String effectorName, Object... args) {
+        return new DslComponent(Scope.THIS, "").effector(effectorName, args);
+    }
+
     /** Returns a {@link Sensor}, looking up the sensor on the context if available and using that,
      * or else defining an untyped (Object) sensor */
     @DslAccessible

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

@@ -20,13 +20,19 @@
 
 import static org.apache.brooklyn.camp.brooklyn.spi.dsl.DslUtils.resolved;
 
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 
+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.Task;
+import org.apache.brooklyn.api.mgmt.TaskAdaptable;
+import org.apache.brooklyn.api.mgmt.TaskFactory;
 import org.apache.brooklyn.api.objs.BrooklynObject;
 import org.apache.brooklyn.api.sensor.AttributeSensor;
 import org.apache.brooklyn.api.sensor.Sensor;
@@ -43,25 +49,32 @@
 import org.apache.brooklyn.core.mgmt.internal.EntityManagerInternal;
 import org.apache.brooklyn.core.sensor.DependentConfiguration;
 import org.apache.brooklyn.core.sensor.Sensors;
+import org.apache.brooklyn.util.collections.MutableMap;
 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.HasSideEffects;
 import org.apache.brooklyn.util.core.task.ImmediateSupplier;
 import org.apache.brooklyn.util.core.task.TaskBuilder;
 import org.apache.brooklyn.util.core.task.Tasks;
 import org.apache.brooklyn.util.exceptions.Exceptions;
 import org.apache.brooklyn.util.groovy.GroovyJavaMethods;
 import org.apache.brooklyn.util.guava.Maybe;
+import org.apache.brooklyn.util.text.StringEscapes.JavaStringEscapes;
 import org.apache.brooklyn.util.text.Strings;
 
 import com.google.common.base.CaseFormat;
 import com.google.common.base.Converter;
+import com.google.common.base.Function;
 import com.google.common.base.Objects;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Predicate;
 import com.google.common.base.Predicates;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.Callables;
 
 public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements DslFunctionSource {
@@ -509,6 +522,119 @@ public String toString() {
         }
     }
 
+    @DslAccessible
+    public BrooklynDslDeferredSupplier<?> effector(final String effectorName) {
+        return new ExecuteEffector(this, effectorName, ImmutableMap.<String, Object>of());
+    }
+    @DslAccessible
+    public BrooklynDslDeferredSupplier<?> effector(final String effectorName, final Map<String, ?> args) {
+        return new ExecuteEffector(this, effectorName, args);
+    }
+    public BrooklynDslDeferredSupplier<?> effector(final String effectorName, Object... args) {
+        return new ExecuteEffector(this, effectorName, ImmutableList.copyOf(args));
+    }
+    protected static class ExecuteEffector extends BrooklynDslDeferredSupplier<Object> implements HasSideEffects {
+        private static final long serialVersionUID = 1740899524088902383L;
+        private final DslComponent component;
+        private final String effectorName;
+        private final Map<String, ?> args;
+        private final List<? extends Object> argList;
+        private Task<?> cachedTask;
+
+        public ExecuteEffector(DslComponent component, String effectorName, Map<String, ?> args) {
+            this.component = Preconditions.checkNotNull(component);
+            this.effectorName = effectorName;
+            this.args = args;
+            this.argList = null;
+        }
+
+        public ExecuteEffector(DslComponent component, String effectorName, List<? extends Object> args) {
+            this.component = Preconditions.checkNotNull(component);
+            this.effectorName = effectorName;
+            this.argList = args;
+            this.args = null;
+        }
+
+        @Override
+        public Maybe<Object> getImmediately() {
+            return Maybe.absent();
+        }
+
+        @SuppressWarnings("unchecked")
+        @Override
+        public Task<Object> newTask() {
+            Entity targetEntity = component.get();
+            Maybe<Effector<?>> targetEffector = targetEntity.getEntityType().getEffectorByName(effectorName);
+            if (targetEffector.isAbsentOrNull()) {
+                throw new IllegalArgumentException("Effector " + effectorName + " not found on entity: " + targetEntity);
+            }
+            synchronized (this) {
+                if (cachedTask == null) {
+                    if (argList == null) {
+                        cachedTask = Entities.invokeEffector(targetEntity, targetEntity, targetEffector.get(), args);
+                    } else {
+                        cachedTask = invokeWithDeferredArgs(targetEntity, targetEffector.get(), argList);
+                    }
+                }
+            }
+            return (Task<Object>) cachedTask;
+        }
+
+        private Task<Object> invokeWithDeferredArgs(final Entity targetEntity, final Effector<?> targetEffector, final List<? extends Object> args) {
+            List<TaskAdaptable<Object>> taskArgs = Lists.newArrayList();
+            for (Object arg : args) {
+                if (arg instanceof TaskAdaptable) {
+                    taskArgs.add((TaskAdaptable<Object>) arg);
+                } else if (arg instanceof TaskFactory) {
+                    taskArgs.add(((TaskFactory<TaskAdaptable<Object>>) arg).newTask());
+                }
+            }
+
+            return DependentConfiguration.transformMultiple(
+                MutableMap.of("displayName", "invoking '"+targetEffector.getName()+"' with "+taskArgs.size()+" task"+(taskArgs.size()!=1?"s":"")), 
+                    new Function<List<Object>, Object>() {
+                        @Override
+                        public Object apply(List<Object> input) {
+                            Iterator<?> tri = input.iterator();
+                            Object[] vv = new Object[args.size()];
+                            int i=0;
+                            for (Object arg : args) {
+                                if (arg instanceof TaskAdaptable || arg instanceof TaskFactory) {
+                                    vv[i] = tri.next();
+                                } else if (arg instanceof DeferredSupplier) {
+                                    vv[i] = ((DeferredSupplier<?>) arg).get();
+                                } else {
+                                    vv[i] = arg;
+                                }
+                                i++;
+                            }
+                            return Entities.invokeEffectorWithArgs(targetEntity, targetEntity, targetEffector, vv);
+                        }
+                    },
+                    taskArgs);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hashCode(component, effectorName);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (this == obj) return true;
+            if (obj == null || getClass() != obj.getClass()) return false;
+            ExecuteEffector that = ExecuteEffector.class.cast(obj);
+            return Objects.equal(this.component, that.component) &&
+                    Objects.equal(this.effectorName, that.effectorName);
+        }
+
+        @Override
+        public String toString() {
+            return (component.scope==Scope.THIS ? "" : component.toString()+".") +
+                "effector("+JavaStringEscapes.wrapJavaString(effectorName)+")";
+        }
+    }
+
     @DslAccessible
     public BrooklynDslDeferredSupplier<?> config(final Object keyNameOrSupplier) {
         return new DslConfigSupplier(this, keyNameOrSupplier);
@@ -763,5 +889,5 @@ public String toString() {
 
         return DslToStringHelpers.component(scopeComponent, remainder);
     }
-    
+
 }

camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/EffectorsYamlIntegrationTest.java

@@ -0,0 +1,98 @@
+/*
+ * 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.camp.brooklyn;
+
+import com.google.common.collect.Iterables;
+import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.core.test.entity.TestEntity;
+import org.apache.brooklyn.entity.software.base.SameServerEntity;
+import org.apache.brooklyn.entity.software.base.VanillaSoftwareProcess;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+
+import static org.apache.brooklyn.core.entity.EntityPredicates.displayNameEqualTo;
+import static org.testng.Assert.assertEquals;
+
+@Test
+public class EffectorsYamlIntegrationTest extends AbstractYamlTest {
+    private static final Logger log = LoggerFactory.getLogger(EffectorsYamlIntegrationTest.class);
+
+    @Test(groups = "Integration")
+    public void testInteractingWithAnotherEntityForStartup() throws Exception {
+
+        final Path tempFile = Files.createTempFile("testInteractingWithAnotherEntityForStartup", ".txt");
+        getLogger().info("Temp file is {}", tempFile.toAbsolutePath());
+
+        try {
+            Entity app = createAndStartApplication(
+                "location: localhost:(name=localhost)",
+                "services:",
+                "- type: " + SameServerEntity.class.getName(),
+                "  brooklyn.children:",
+                "  - type: " + TestEntity.class.getName(),
+                "    id: testEntity",
+                "    name: testEntity",
+                "    brooklyn.initializers:",
+                "    - type: org.apache.brooklyn.core.sensor.ssh.SshCommandSensor",
+                "      brooklyn.config:",
+                "        name: greeting",
+                "        period: 2s",
+                "        command: |",
+                "          echo hello world",
+                "  - type: " + VanillaSoftwareProcess.class.getName(),
+                "    id: consumerEntity",
+                "    brooklyn.config:",
+                "      install.latch: $brooklyn:entity(\"testEntity\").attributeWhenReady(\"service.isUp\")",
+                "      launch.command: while true; do sleep 3600 ; done & echo $! > ${PID_FILE}",
+                "      shell.env:",
+                "        RESPONSE: $brooklyn:entity(\"testEntity\").effector(\"identityEffector\", $brooklyn:entity(\"testEntity\").attributeWhenReady(\"greeting\"))",
+                "      post.launch.command: echo ${RESPONSE} > " + tempFile.toAbsolutePath()
+            );
+            waitForApplicationTasks(app);
+
+            final String contents = new String(Files.readAllBytes(tempFile)).trim();
+            assertEquals(contents, "hello world", "file contents: " + contents);
+
+        } finally {
+            Files.delete(tempFile);
+        }
+    }
+
+
+    private void assertCallHistory(TestEntity testEntity, String... expectedCalls) {
+        List<String> callHistory = testEntity.getCallHistory();
+        Assert.assertEquals(callHistory.size(), expectedCalls.length, "history = " + callHistory);
+        int c = 0;
+        for (String expected : expectedCalls) {
+            Assert.assertEquals(callHistory.get(c++), expected, "history = " + callHistory);
+        }
+    }
+
+    @Override
+    protected Logger getLogger() {
+        return log;
+    }
+
+}