Add array access support for mustache templates

The default mustache template that is supported by elasticsearch doesn't support array/list access. This poses a real limitation for watcher as with `search` input, the hits are returned as an array/list. To bypass this limitation, an extra (tedious) step is required in order to transform the hits to a data structure that is supported by mustache.

This commit adds a new mustache script engine - `xmustache` to elasticsearch that supports array/list access in the form of `array.X` where `X` is the index into the array/list. This enables accessing the search results without using a transform. The following example will fetch the `"key"` field of the 3rd hit in the search result: `ctx.payload.hits.hits.3._source.key`.

This array/list support will be added to elasticsearch, but it'll only be available in later versions. For now, the default template in watcher will therefore be `xmustache`.

Added docs for templates

Fixes elastic/elasticsearch#230

Original commit: elastic/x-pack-elasticsearch@b09cad7f8b

src/main/java/org/elasticsearch/watcher/support/template/MustacheTemplateEngine.java

@@ -10,6 +10,7 @@ import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.script.ExecutableScript;
+import org.elasticsearch.script.mustache.MustacheScriptEngineService;
 import org.elasticsearch.watcher.support.init.proxy.ScriptServiceProxy;
 
 import java.util.HashMap;
@@ -33,7 +34,7 @@ public class MustacheTemplateEngine extends AbstractComponent implements Templat
         Map<String, Object> mergedModel = new HashMap<>();
         mergedModel.putAll(template.getParams());
         mergedModel.putAll(model);
-        ExecutableScript executable = service.executable("mustache", template.getText(), template.getType(), mergedModel);
+        ExecutableScript executable = service.executable(MustacheScriptEngineService.NAME, template.getText(), template.getType(), mergedModel);
         Object result = executable.run();
         if (result instanceof BytesReference) {
             return ((BytesReference) result).toUtf8();

src/main/java/org/elasticsearch/watcher/support/template/TemplateModule.java

@@ -6,15 +6,27 @@
 package org.elasticsearch.watcher.support.template;
 
 import org.elasticsearch.common.inject.AbstractModule;
+import org.elasticsearch.common.inject.Module;
+import org.elasticsearch.common.inject.PreProcessModule;
+import org.elasticsearch.script.ScriptModule;
+import org.elasticsearch.watcher.support.template.xmustache.XMustacheScriptEngineService;
+import org.elasticsearch.watcher.support.template.xmustache.XMustacheTemplateEngine;
 
 /**
  *
  */
-public class TemplateModule extends AbstractModule {
+public class TemplateModule extends AbstractModule implements PreProcessModule {
+
+    @Override
+    public void processModule(Module module) {
+        if (module instanceof ScriptModule) {
+            ((ScriptModule) module).addScriptEngine(XMustacheScriptEngineService.class);
+        }
+    }
 
     @Override
     protected void configure() {
-        bind(MustacheTemplateEngine.class).asEagerSingleton();
-        bind(TemplateEngine.class).to(MustacheTemplateEngine.class);
+        bind(XMustacheTemplateEngine.class).asEagerSingleton();
+        bind(TemplateEngine.class).to(XMustacheTemplateEngine.class);
     }
 }

src/main/java/org/elasticsearch/watcher/support/template/xmustache/XMustacheFactory.java

@@ -0,0 +1,202 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.support.template.xmustache;
+
+import org.elasticsearch.common.collect.Iterables;
+import org.elasticsearch.common.mustache.DefaultMustacheFactory;
+import org.elasticsearch.common.mustache.MustacheException;
+import org.elasticsearch.common.mustache.reflect.ReflectionObjectHandler;
+
+import java.io.IOException;
+import java.io.Writer;
+import java.lang.reflect.Array;
+import java.util.*;
+
+/**
+ * An extension to elasticsearch's {@code JsonEscapingMustacheFactory} that on top of applying json
+ * escapes it also enables support for navigating arrays using `array.X` notation (where `X` is the index
+ * of the element in the array).
+ */
+public class XMustacheFactory extends DefaultMustacheFactory {
+
+    public XMustacheFactory() {
+        setObjectHandler(new ReflectionObjectHandler() {
+            @Override
+            public Object coerce(Object object) {
+                if (object != null) {
+                    if (object.getClass().isArray()) {
+                        return new ArrayMap(object);
+                    } else if (object instanceof Collection) {
+                        return new CollectionMap((Collection) object);
+                    }
+                }
+                return super.coerce(object);
+            }
+        });
+    }
+
+    @Override
+    public void encode(String value, Writer writer) {
+        try {
+            escape(value, writer);
+        } catch (IOException e) {
+            throw new MustacheException("Failed to encode value: " + value);
+        }
+    }
+
+    public static Writer escape(String value, Writer writer) throws IOException {
+        for (int i = 0; i < value.length(); i++) {
+            final char character = value.charAt(i);
+            if (isEscapeChar(character)) {
+                writer.write('\\');
+            }
+            writer.write(character);
+        }
+        return writer;
+    }
+
+    public static boolean isEscapeChar(char c) {
+        switch (c) {
+            case '\b':
+            case '\f':
+            case '\n':
+            case '\r':
+            case '"':
+            case '\\':
+            case '\u000B': // vertical tab
+            case '\t':
+                return true;
+        }
+        return false;
+    }
+
+
+    static class ArrayMap extends AbstractMap<Object, Object> implements Iterable<Object> {
+
+        private final Object array;
+
+        public ArrayMap(Object array) {
+            this.array = array;
+        }
+
+        @Override
+        public Object get(Object key) {
+            if (key instanceof Number) {
+                return Array.get(array, ((Number) key).intValue());
+            }
+            try {
+                int index = Integer.parseInt(key.toString());
+                return Array.get(array, index);
+            } catch (NumberFormatException nfe) {
+                // if it's not a number it is as if the key doesn't exist
+                return null;
+            }
+        }
+
+        @Override
+        public boolean containsKey(Object key) {
+            return get(key) != null;
+        }
+
+        @Override
+        public Set<Entry<Object, Object>> entrySet() {
+            int length = Array.getLength(array);
+            Map<Object, Object> map = new HashMap<>(length);
+            for (int i = 0; i < length; i++) {
+                map.put(i, Array.get(array, i));
+            }
+            return map.entrySet();
+        }
+
+        /**
+         * Returns an iterator over a set of elements of type T.
+         *
+         * @return an Iterator.
+         */
+        @Override
+        public Iterator<Object> iterator() {
+            return new Iter(array);
+        }
+
+        static class Iter implements Iterator<Object> {
+
+            private final Object array;
+            private final int length;
+            private int index;
+
+            public Iter(Object array) {
+                this.array = array;
+                this.length = Array.getLength(array);
+                this.index = 0;
+            }
+
+            @Override
+            public boolean hasNext() {
+                return index < length;
+            }
+
+            @Override
+            public Object next() {
+                return Array.get(array, index++);
+            }
+
+            @Override
+            public void remove() {
+                throw new UnsupportedOperationException("array iterator does not support removing elements");
+            }
+        }
+    }
+
+    static class CollectionMap extends AbstractMap<Object, Object> implements Iterable<Object> {
+
+        private final Collection col;
+
+        public CollectionMap(Collection col) {
+            this.col = col;
+        }
+
+        @Override
+        public Object get(Object key) {
+            if (key instanceof Number) {
+                return Iterables.get(col, ((Number) key).intValue());
+            }
+            try {
+                int index = Integer.parseInt(key.toString());
+                return Iterables.get(col, index);
+            } catch (NumberFormatException nfe) {
+                // if it's not a number it is as if the key doesn't exist
+                return null;
+            }
+        }
+
+        @Override
+        public boolean containsKey(Object key) {
+            return get(key) != null;
+        }
+
+        @Override
+        public Set<Entry<Object, Object>> entrySet() {
+            Map<Object, Object> map = new HashMap<>(col.size());
+            int i = 0;
+            for (Object item : col) {
+                map.put(i++, item);
+            }
+            return map.entrySet();
+        }
+
+        /**
+         * Returns an iterator over a set of elements of type T.
+         *
+         * @return an Iterator.
+         */
+        @Override
+        public Iterator<Object> iterator() {
+            return col.iterator();
+        }
+    }
+
+
+}

src/main/java/org/elasticsearch/watcher/support/template/xmustache/XMustacheScriptEngineService.java

@@ -0,0 +1,191 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.support.template.xmustache;
+
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.component.AbstractComponent;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.io.FastStringReader;
+import org.elasticsearch.common.io.UTF8StreamWriter;
+import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.common.mustache.Mustache;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.script.CompiledScript;
+import org.elasticsearch.script.ExecutableScript;
+import org.elasticsearch.script.ScriptEngineService;
+import org.elasticsearch.script.SearchScript;
+import org.elasticsearch.search.lookup.SearchLookup;
+
+import java.io.IOException;
+import java.lang.ref.SoftReference;
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ *
+ */
+public class XMustacheScriptEngineService extends AbstractComponent implements ScriptEngineService {
+
+    public static final String NAME = "xmustache";
+
+    /**
+     * @param settings automatically wired by Guice.
+     * */
+    @Inject
+    public XMustacheScriptEngineService(Settings settings) {
+        super(settings);
+    }
+
+    /**
+     * Compile a template string to (in this case) a Mustache object than can
+     * later be re-used for execution to fill in missing parameter values.
+     *
+     * @param template
+     *            a string representing the template to compile.
+     * @return a compiled template object for later execution.
+     * */
+    @Override
+    public Object compile(String template) {
+        /** Factory to generate Mustache objects from. */
+        return (new XMustacheFactory()).compile(new FastStringReader(template), "query-template");
+    }
+
+    /**
+     * Execute a compiled template object (as retrieved from the compile method)
+     * and fill potential place holders with the variables given.
+     *
+     * @param template
+     *            compiled template object.
+     * @param vars
+     *            map of variables to use during substitution.
+     *
+     * @return the processed string with all given variables substitued.
+     * */
+    @Override
+    public Object execute(Object template, Map<String, Object> vars) {
+        BytesStreamOutput result = new BytesStreamOutput();
+        UTF8StreamWriter writer = utf8StreamWriter().setOutput(result);
+        ((Mustache) template).execute(writer, vars);
+        try {
+            writer.flush();
+        } catch (IOException e) {
+            logger.error("Could not execute query template (failed to flush writer): ", e);
+        } finally {
+            try {
+                writer.close();
+            } catch (IOException e) {
+                logger.error("Could not execute query template (failed to close writer): ", e);
+            }
+        }
+        return result.bytes();
+    }
+
+    @Override
+    public String[] types() {
+        return new String[] { NAME };
+    }
+
+    @Override
+    public String[] extensions() {
+        return new String[] { NAME };
+    }
+
+    @Override
+    public boolean sandboxed() {
+        return true;
+    }
+
+    @Override
+    public ExecutableScript executable(Object mustache,
+                                       @Nullable Map<String, Object> vars) {
+        return new MustacheExecutableScript((Mustache) mustache, vars);
+    }
+
+    @Override
+    public SearchScript search(Object compiledScript, SearchLookup lookup,
+                               @Nullable Map<String, Object> vars) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Object unwrap(Object value) {
+        return value;
+    }
+
+    @Override
+    public void close() {
+        // Nothing to do here
+    }
+
+    @Override
+    public void scriptRemoved(CompiledScript script) {
+        // Nothing to do here
+    }
+
+    /**
+     * Used at query execution time by script service in order to execute a query template.
+     * */
+    private class MustacheExecutableScript implements ExecutableScript {
+        /** Compiled template object. */
+        private Mustache mustache;
+        /** Parameters to fill above object with. */
+        private Map<String, Object> vars;
+
+        /**
+         * @param mustache the compiled template object
+         * @param vars the parameters to fill above object with
+         **/
+        public MustacheExecutableScript(Mustache mustache,
+                                        Map<String, Object> vars) {
+            this.mustache = mustache;
+            this.vars = vars == null ? Collections.<String, Object>emptyMap() : vars;
+        }
+
+        @Override
+        public void setNextVar(String name, Object value) {
+            this.vars.put(name, value);
+        }
+
+        @Override
+        public Object run() {
+            BytesStreamOutput result = new BytesStreamOutput();
+            UTF8StreamWriter writer = utf8StreamWriter().setOutput(result);
+            mustache.execute(writer, vars);
+            try {
+                writer.flush();
+            } catch (IOException e) {
+                logger.error("Could not execute query template (failed to flush writer): ", e);
+            } finally {
+                try {
+                    writer.close();
+                } catch (IOException e) {
+                    logger.error("Could not execute query template (failed to close writer): ", e);
+                }
+            }
+            return result.bytes();
+        }
+
+        @Override
+        public Object unwrap(Object value) {
+            return value;
+        }
+    }
+
+    /** Thread local UTF8StreamWriter to store template execution results in, thread local to save object creation.*/
+    private static ThreadLocal<SoftReference<UTF8StreamWriter>> utf8StreamWriter = new ThreadLocal<>();
+
+    /** If exists, reset and return, otherwise create, reset and return a writer.*/
+    private static UTF8StreamWriter utf8StreamWriter() {
+        SoftReference<UTF8StreamWriter> ref = utf8StreamWriter.get();
+        UTF8StreamWriter writer = (ref == null) ? null : ref.get();
+        if (writer == null) {
+            writer = new UTF8StreamWriter(1024 * 4);
+            utf8StreamWriter.set(new SoftReference<>(writer));
+        }
+        writer.reset();
+        return writer;
+    }
+}

src/main/java/org/elasticsearch/watcher/support/template/xmustache/XMustacheTemplateEngine.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.support.template.xmustache;
+
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.component.AbstractComponent;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.script.ExecutableScript;
+import org.elasticsearch.watcher.support.init.proxy.ScriptServiceProxy;
+import org.elasticsearch.watcher.support.template.Template;
+import org.elasticsearch.watcher.support.template.TemplateEngine;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ *
+ */
+public class XMustacheTemplateEngine extends AbstractComponent implements TemplateEngine {
+
+    private final ScriptServiceProxy service;
+
+    @Inject
+    public XMustacheTemplateEngine(Settings settings, ScriptServiceProxy service) {
+        super(settings);
+        this.service = service;
+    }
+
+    @Override
+    public String render(Template template, Map<String, Object> model) {
+        Map<String, Object> mergedModel = new HashMap<>();
+        mergedModel.putAll(template.getParams());
+        mergedModel.putAll(model);
+        ExecutableScript executable = service.executable(XMustacheScriptEngineService.NAME, template.getText(), template.getType(), mergedModel);
+        Object result = executable.run();
+        if (result instanceof BytesReference) {
+            return ((BytesReference) result).toUtf8();
+        }
+        return result.toString();
+    }
+}

src/test/java/org/elasticsearch/watcher/actions/webhook/WebhookActionTests.java

@@ -20,7 +20,6 @@ import org.elasticsearch.env.Environment;
 import org.elasticsearch.node.settings.NodeSettingsService;
 import org.elasticsearch.script.ScriptEngineService;
 import org.elasticsearch.script.ScriptService;
-import org.elasticsearch.script.mustache.MustacheScriptEngineService;
 import org.elasticsearch.test.ElasticsearchTestCase;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.watcher.ResourceWatcherService;
@@ -35,9 +34,10 @@ import org.elasticsearch.watcher.support.http.auth.HttpAuth;
 import org.elasticsearch.watcher.support.http.auth.HttpAuthRegistry;
 import org.elasticsearch.watcher.support.init.proxy.ClientProxy;
 import org.elasticsearch.watcher.support.init.proxy.ScriptServiceProxy;
-import org.elasticsearch.watcher.support.template.MustacheTemplateEngine;
 import org.elasticsearch.watcher.support.template.Template;
 import org.elasticsearch.watcher.support.template.TemplateEngine;
+import org.elasticsearch.watcher.support.template.xmustache.XMustacheScriptEngineService;
+import org.elasticsearch.watcher.support.template.xmustache.XMustacheTemplateEngine;
 import org.elasticsearch.watcher.test.WatcherTestUtils;
 import org.elasticsearch.watcher.trigger.schedule.ScheduleTriggerEvent;
 import org.elasticsearch.watcher.watch.Payload;
@@ -83,11 +83,11 @@ public class WebhookActionTests extends ElasticsearchTestCase {
     public void init() throws IOException {
         tp = new ThreadPool(ThreadPool.Names.SAME);
         Settings settings = ImmutableSettings.EMPTY;
-        MustacheScriptEngineService mustacheScriptEngineService = new MustacheScriptEngineService(settings);
+        XMustacheScriptEngineService mustacheScriptEngineService = new XMustacheScriptEngineService(settings);
         Set<ScriptEngineService> engineServiceSet = new HashSet<>();
         engineServiceSet.add(mustacheScriptEngineService);
         scriptService = ScriptServiceProxy.of(new ScriptService(settings, new Environment(), engineServiceSet, new ResourceWatcherService(settings, tp), new NodeSettingsService(settings)));
-        templateEngine = new MustacheTemplateEngine(settings, scriptService);
+        templateEngine = new XMustacheTemplateEngine(settings, scriptService);
         testBody = new Template(TEST_BODY_STRING );
         testPath = new Template(TEST_PATH_STRING);
         authRegistry = new HttpAuthRegistry(ImmutableMap.of("basic", (HttpAuth.Parser) new BasicAuth.Parser()));

src/test/java/org/elasticsearch/watcher/support/template/TemplateTests.java

@@ -15,6 +15,7 @@ import org.elasticsearch.script.ExecutableScript;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.test.ElasticsearchTestCase;
 import org.elasticsearch.watcher.support.init.proxy.ScriptServiceProxy;
+import org.elasticsearch.watcher.support.template.xmustache.XMustacheTemplateEngine;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -33,13 +34,13 @@ public class TemplateTests extends ElasticsearchTestCase {
     private ScriptServiceProxy proxy;
     private TemplateEngine engine;
     private ExecutableScript script;
-    private final String lang = "mustache";
+    private final String lang = "xmustache";
 
     @Before
     public void init() throws Exception {
         proxy = mock(ScriptServiceProxy.class);
         script = mock(ExecutableScript.class);
-        engine = new MustacheTemplateEngine(ImmutableSettings.EMPTY, proxy);
+        engine = new XMustacheTemplateEngine(ImmutableSettings.EMPTY, proxy);
     }
 
     @Test

src/test/java/org/elasticsearch/watcher/support/template/xmustache/XMustacheScriptEngineTests.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.support.template.xmustache;
+
+import com.carrotsearch.randomizedtesting.generators.RandomPicks;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.test.ElasticsearchTestCase;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.io.StringWriter;
+import java.nio.charset.Charset;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.hamcrest.Matchers.equalTo;
+
+/**
+ *
+ */
+public class XMustacheScriptEngineTests extends ElasticsearchTestCase {
+
+    private XMustacheScriptEngineService engine;
+
+    @Before
+    public void setup() {
+        engine = new XMustacheScriptEngineService(ImmutableSettings.Builder.EMPTY_SETTINGS);
+    }
+
+    @Test
+    public void testSimpleParameterReplace() {
+        {
+            String template = "GET _search {\"query\": " + "{\"boosting\": {" + "\"positive\": {\"match\": {\"body\": \"gift\"}},"
+                    + "\"negative\": {\"term\": {\"body\": {\"value\": \"solr\"}" + "}}, \"negative_boost\": {{boost_val}} } }}";
+            Map<String, Object> vars = new HashMap<>();
+            vars.put("boost_val", "0.3");
+            BytesReference o = (BytesReference) engine.execute(engine.compile(template), vars);
+            assertEquals("GET _search {\"query\": {\"boosting\": {\"positive\": {\"match\": {\"body\": \"gift\"}},"
+                            + "\"negative\": {\"term\": {\"body\": {\"value\": \"solr\"}}}, \"negative_boost\": 0.3 } }}",
+                    new String(o.toBytes(), Charset.forName("UTF-8")));
+        }
+        {
+            String template = "GET _search {\"query\": " + "{\"boosting\": {" + "\"positive\": {\"match\": {\"body\": \"gift\"}},"
+                    + "\"negative\": {\"term\": {\"body\": {\"value\": \"{{body_val}}\"}" + "}}, \"negative_boost\": {{boost_val}} } }}";
+            Map<String, Object> vars = new HashMap<>();
+            vars.put("boost_val", "0.3");
+            vars.put("body_val", "\"quick brown\"");
+            BytesReference o = (BytesReference) engine.execute(engine.compile(template), vars);
+            assertEquals("GET _search {\"query\": {\"boosting\": {\"positive\": {\"match\": {\"body\": \"gift\"}},"
+                            + "\"negative\": {\"term\": {\"body\": {\"value\": \"\\\"quick brown\\\"\"}}}, \"negative_boost\": 0.3 } }}",
+                    new String(o.toBytes(), Charset.forName("UTF-8")));
+        }
+    }
+
+    @Test
+    public void testEscapeJson() throws IOException {
+        {
+            StringWriter writer = new StringWriter();
+            XMustacheFactory.escape("hello \n world", writer);
+            assertThat(writer.toString(), equalTo("hello \\\n world"));
+        }
+        {
+            StringWriter writer = new StringWriter();
+            XMustacheFactory.escape("\n", writer);
+            assertThat(writer.toString(), equalTo("\\\n"));
+        }
+
+        Character[] specialChars = new Character[]{'\f', '\n', '\r', '"', '\\', (char) 11, '\t', '\b' };
+        int iters = scaledRandomIntBetween(100, 1000);
+        for (int i = 0; i < iters; i++) {
+            int rounds = scaledRandomIntBetween(1, 20);
+            StringWriter escaped = new StringWriter();
+            StringWriter writer = new StringWriter();
+            for (int j = 0; j < rounds; j++) {
+                String s = getChars();
+                writer.write(s);
+                escaped.write(s);
+                char c = RandomPicks.randomFrom(getRandom(), specialChars);
+                writer.append(c);
+                escaped.append('\\');
+                escaped.append(c);
+            }
+            StringWriter target = new StringWriter();
+            assertThat(escaped.toString(), equalTo(XMustacheFactory.escape(writer.toString(), target).toString()));
+        }
+    }
+
+    private String getChars() {
+        String string = randomRealisticUnicodeOfCodepointLengthBetween(0, 10);
+        for (int i = 0; i < string.length(); i++) {
+            if (XMustacheFactory.isEscapeChar(string.charAt(i))) {
+                return string.substring(0, i);
+            }
+        }
+        return string;
+    }
+
+}

src/test/java/org/elasticsearch/watcher/support/template/xmustache/XMustacheTests.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.support.template.xmustache;
+
+import com.carrotsearch.ant.tasks.junit4.dependencies.com.google.common.collect.ImmutableList;
+import com.carrotsearch.randomizedtesting.annotations.Repeat;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.ImmutableMap;
+import org.elasticsearch.common.collect.ImmutableSet;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.script.ScriptEngineService;
+import org.elasticsearch.test.ElasticsearchTestCase;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.hamcrest.Matchers.*;
+
+/**
+ *
+ */
+public class XMustacheTests extends ElasticsearchTestCase {
+
+    private ScriptEngineService engine;
+
+    @Before
+    public void init() throws Exception {
+        engine = new XMustacheScriptEngineService(ImmutableSettings.EMPTY);
+    }
+
+    @Test @Repeat(iterations = 10)
+    public void testArrayAccess() throws Exception {
+        String template = "{{data.0}} {{data.1}}";
+        Object mustache = engine.compile(template);
+        Map<String, Object> vars = new HashMap<>();
+        Object data = randomFrom(
+                new String[] { "foo", "bar" },
+                ImmutableList.of("foo", "bar"),
+                ImmutableSet.of("foo", "bar"));
+        vars.put("data", data);
+        Object output = engine.execute(mustache, vars);
+        assertThat(output, notNullValue());
+        assertThat(output, instanceOf(BytesReference.class));
+        BytesReference bytes = (BytesReference) output;
+        assertThat(bytes.toUtf8(), equalTo("foo bar"));
+    }
+
+    @Test @Repeat(iterations = 10)
+    public void testArrayInArrayAccess() throws Exception {
+        String template = "{{data.0.0}} {{data.0.1}}";
+        Object mustache = engine.compile(template);
+        Map<String, Object> vars = new HashMap<>();
+        Object data = randomFrom(
+                new String[][] { new String[] {"foo", "bar" }},
+                ImmutableList.of(new String[] {"foo", "bar" }),
+                ImmutableSet.of(new String[] {"foo", "bar" })
+        );
+        vars.put("data", data);
+        Object output = engine.execute(mustache, vars);
+        assertThat(output, notNullValue());
+        assertThat(output, instanceOf(BytesReference.class));
+        BytesReference bytes = (BytesReference) output;
+        assertThat(bytes.toUtf8(), equalTo("foo bar"));
+    }
+
+    @Test @Repeat(iterations = 10)
+    public void testMapInArrayAccess() throws Exception {
+        String template = "{{data.0.key}} {{data.1.key}}";
+        Object mustache = engine.compile(template);
+        Map<String, Object> vars = new HashMap<>();
+        Object data = randomFrom(
+                new Map[] { ImmutableMap.<String, Object>of("key", "foo"), ImmutableMap.<String, Object>of("key", "bar") },
+                ImmutableList.of(ImmutableMap.<String, Object>of("key", "foo"), ImmutableMap.<String, Object>of("key", "bar")),
+                ImmutableSet.of(ImmutableMap.<String, Object>of("key", "foo"), ImmutableMap.<String, Object>of("key", "bar")));
+        vars.put("data", data);
+        Object output = engine.execute(mustache, vars);
+        assertThat(output, notNullValue());
+        assertThat(output, instanceOf(BytesReference.class));
+        BytesReference bytes = (BytesReference) output;
+        assertThat(bytes.toUtf8(), equalTo("foo bar"));
+    }
+}

src/test/java/org/elasticsearch/watcher/test/WatcherTestUtils.java

@@ -40,7 +40,7 @@ import org.elasticsearch.watcher.support.http.HttpMethod;
 import org.elasticsearch.watcher.support.http.HttpRequestTemplate;
 import org.elasticsearch.watcher.support.init.proxy.ClientProxy;
 import org.elasticsearch.watcher.support.init.proxy.ScriptServiceProxy;
-import org.elasticsearch.watcher.support.template.MustacheTemplateEngine;
+import org.elasticsearch.watcher.support.template.xmustache.XMustacheTemplateEngine;
 import org.elasticsearch.watcher.support.template.Template;
 import org.elasticsearch.watcher.support.template.TemplateEngine;
 import org.elasticsearch.watcher.transform.SearchTransform;
@@ -143,7 +143,7 @@ public final class WatcherTestUtils {
         Template body = new Template("{{ctx.watch_id}} executed with {{ctx.payload.response.hits.total_hits}} hits");
         httpRequest.body(body);
 
-        TemplateEngine engine = new MustacheTemplateEngine(ImmutableSettings.EMPTY, scriptService);
+        TemplateEngine engine = new XMustacheTemplateEngine(ImmutableSettings.EMPTY, scriptService);
 
         actions.add(new ActionWrapper("_webhook", new ExecutableWebhookAction(new WebhookAction(httpRequest.build()), logger, httpClient, engine)));
 
@@ -155,7 +155,7 @@ public final class WatcherTestUtils {
                 .to(to)
                 .build();
 
-        TemplateEngine templateEngine = new MustacheTemplateEngine(ImmutableSettings.EMPTY, scriptService);
+        TemplateEngine templateEngine = new XMustacheTemplateEngine(ImmutableSettings.EMPTY, scriptService);
 
         Authentication auth = new Authentication("testname", "testpassword".toCharArray());
 

src/test/java/org/elasticsearch/watcher/test/integration/EmailActionIntegrationTests.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.test.integration;
+
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.watcher.actions.email.service.EmailTemplate;
+import org.elasticsearch.watcher.actions.email.service.support.EmailServer;
+import org.elasticsearch.watcher.client.WatcherClient;
+import org.elasticsearch.watcher.test.AbstractWatcherIntegrationTests;
+import org.elasticsearch.watcher.trigger.schedule.IntervalSchedule;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import javax.mail.internet.MimeMessage;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.elasticsearch.index.query.QueryBuilders.termQuery;
+import static org.elasticsearch.search.builder.SearchSourceBuilder.searchSource;
+import static org.elasticsearch.watcher.actions.ActionBuilders.emailAction;
+import static org.elasticsearch.watcher.client.WatchSourceBuilders.watchBuilder;
+import static org.elasticsearch.watcher.condition.ConditionBuilders.scriptCondition;
+import static org.elasticsearch.watcher.input.InputBuilders.searchInput;
+import static org.elasticsearch.watcher.test.WatcherTestUtils.newInputSearchRequest;
+import static org.elasticsearch.watcher.trigger.TriggerBuilders.schedule;
+import static org.elasticsearch.watcher.trigger.schedule.Schedules.interval;
+import static org.hamcrest.Matchers.equalTo;
+
+/**
+ *
+ */
+public class EmailActionIntegrationTests extends AbstractWatcherIntegrationTests {
+
+    static final String USERNAME = "_user";
+    static final String PASSWORD = "_passwd";
+
+    private EmailServer server;
+
+    @Before
+    public void init() throws Exception {
+        server = new EmailServer("localhost", 2500, USERNAME, PASSWORD);
+        server.start();
+    }
+
+    @After
+    public void cleanup() throws Exception {
+        server.stop();
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return ImmutableSettings.builder()
+                .put(super.nodeSettings(nodeOrdinal))
+                .put("watcher.actions.email.service.account.test.smtp.auth", true)
+                .put("watcher.actions.email.service.account.test.smtp.user", USERNAME)
+                .put("watcher.actions.email.service.account.test.smtp.password", PASSWORD)
+                .put("watcher.actions.email.service.account.test.smtp.port", 2500)
+                .put("watcher.actions.email.service.account.test.smtp.host", "localhost")
+                .build();
+    }
+
+    @Test
+    public void testArrayAccess() throws Exception {
+        final CountDownLatch latch = new CountDownLatch(1);
+        EmailServer.Listener.Handle handle = server.addListener(new EmailServer.Listener() {
+            @Override
+            public void on(MimeMessage message) throws Exception {
+                assertThat(message.getSubject(), equalTo("value"));
+                latch.countDown();
+            }
+        });
+
+        WatcherClient watcherClient = watcherClient();
+        createIndex("idx");
+        // Have a sample document in the index, the watch is going to evaluate
+        client().prepareIndex("idx", "type").setSource("field", "value").get();
+        refresh();
+        SearchRequest searchRequest = newInputSearchRequest("idx").source(searchSource().query(termQuery("field", "value")));
+        watcherClient.preparePutWatch("_id")
+                .setSource(watchBuilder()
+                        .trigger(schedule(interval(5, IntervalSchedule.Interval.Unit.SECONDS)))
+                        .input(searchInput(searchRequest))
+                        .condition(scriptCondition("ctx.payload.hits.total > 0"))
+                        .addAction("_index", emailAction(EmailTemplate.builder().from("_from").to("_to")
+                                .subject("{{ctx.payload.hits.hits.0._source.field}}"))))
+                        .get();
+
+        if (timeWarped()) {
+            timeWarp().scheduler().trigger("_id");
+            refresh();
+        }
+
+        assertWatchWithMinimumPerformedActionsCount("_id", 1);
+
+        if (!latch.await(5, TimeUnit.SECONDS)) {
+            fail("waited too long for email to be received");
+        }
+    }
+}