Introduced XContentSource

An absraction aover XContent byte reference that:

- implements `ToXContent`
- can resolve values given a dot-notation path
- Used in `GetWatchResponse` and `ExecuteWatchResponse`

Also

- Moved `WatchExecutionResult` to the `execution` package

Original commit: elastic/x-pack-elasticsearch@0b41f53f38

src/main/java/org/elasticsearch/watcher/client/WatchSourceBuilder.java

@@ -19,6 +19,7 @@ import org.elasticsearch.watcher.condition.Condition;
 import org.elasticsearch.watcher.condition.always.AlwaysCondition;
 import org.elasticsearch.watcher.input.Input;
 import org.elasticsearch.watcher.input.none.NoneInput;
+import org.elasticsearch.watcher.support.xcontent.XContentSource;
 import org.elasticsearch.watcher.transform.Transform;
 import org.elasticsearch.watcher.trigger.Trigger;
 import org.elasticsearch.watcher.watch.Watch;
@@ -27,6 +28,8 @@ import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+
 /**
  *
  */
@@ -110,6 +113,10 @@ public class WatchSourceBuilder implements ToXContent {
         return this;
     }
 
+    public XContentSource build() throws IOException {
+        return new XContentSource(toXContent(jsonBuilder(), ToXContent.EMPTY_PARAMS).bytes());
+    }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();

src/main/java/org/elasticsearch/watcher/execution/ExecutionService.java

@@ -26,7 +26,6 @@ import org.elasticsearch.watcher.transform.ExecutableTransform;
 import org.elasticsearch.watcher.transform.Transform;
 import org.elasticsearch.watcher.trigger.TriggerEvent;
 import org.elasticsearch.watcher.watch.Watch;
-import org.elasticsearch.watcher.watch.WatchExecutionResult;
 import org.elasticsearch.watcher.watch.WatchLockService;
 import org.elasticsearch.watcher.watch.WatchStore;
 

src/main/java/org/elasticsearch/watcher/execution/WatchExecutionContext.java

@@ -15,7 +15,6 @@ import org.elasticsearch.watcher.transform.Transform;
 import org.elasticsearch.watcher.trigger.TriggerEvent;
 import org.elasticsearch.watcher.watch.Payload;
 import org.elasticsearch.watcher.watch.Watch;
-import org.elasticsearch.watcher.watch.WatchExecutionResult;
 
 import java.util.HashMap;
 import java.util.Map;

src/main/java/org/elasticsearch/watcher/execution/WatchExecutionResult.java

@@ -3,7 +3,7 @@
  * 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.watch;
+package org.elasticsearch.watcher.execution;
 
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.ParseField;
@@ -17,8 +17,6 @@ import org.elasticsearch.watcher.actions.ActionWrapper;
 import org.elasticsearch.watcher.actions.ExecutableActions;
 import org.elasticsearch.watcher.condition.Condition;
 import org.elasticsearch.watcher.condition.ConditionRegistry;
-import org.elasticsearch.watcher.execution.WatchExecutionContext;
-import org.elasticsearch.watcher.execution.Wid;
 import org.elasticsearch.watcher.input.Input;
 import org.elasticsearch.watcher.input.InputRegistry;
 import org.elasticsearch.watcher.support.WatcherDateUtils;

src/main/java/org/elasticsearch/watcher/history/WatchRecord.java

@@ -28,7 +28,7 @@ import org.elasticsearch.watcher.transform.TransformRegistry;
 import org.elasticsearch.watcher.trigger.TriggerEvent;
 import org.elasticsearch.watcher.trigger.TriggerService;
 import org.elasticsearch.watcher.watch.Watch;
-import org.elasticsearch.watcher.watch.WatchExecutionResult;
+import org.elasticsearch.watcher.execution.WatchExecutionResult;
 
 import java.io.IOException;
 import java.util.Locale;

src/main/java/org/elasticsearch/watcher/rest/action/RestExecuteWatchAction.java

@@ -48,7 +48,7 @@ public class RestExecuteWatchAction extends WatcherRestHandler {
         client.executeWatch(executeWatchRequest, new RestBuilderListener<ExecuteWatchResponse>(channel) {
             @Override
             public RestResponse buildResponse(ExecuteWatchResponse response, XContentBuilder builder) throws Exception {
-                builder.value(response.getWatchRecordAsMap());
+                builder.value(response.getSource());
                 return new BytesRestResponse(RestStatus.OK, builder);
             }
         });

src/main/java/org/elasticsearch/watcher/rest/action/RestGetWatchAction.java

@@ -8,6 +8,7 @@ package org.elasticsearch.watcher.rest.action;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.*;
 import org.elasticsearch.rest.action.support.RestBuilderListener;
@@ -38,7 +39,7 @@ public class RestGetWatchAction extends WatcherRestHandler {
                         .field("found", response.isFound())
                         .field("_id", response.getId())
                         .field("_version", response.getVersion())
-                        .field("watch", response.getSourceAsMap())
+                        .field("watch", response.getSource(), ToXContent.EMPTY_PARAMS)
                         .endObject();
 
                 RestStatus status = response.isFound() ? OK : NOT_FOUND;

src/main/java/org/elasticsearch/watcher/support/xcontent/XContentSource.java

@@ -0,0 +1,88 @@
+/*
+ * 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.xcontent;
+
+import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.*;
+import org.elasticsearch.common.xcontent.support.XContentMapValues;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * Encapsulates the xcontent source
+ */
+public class XContentSource implements ToXContent {
+
+    private final BytesReference bytes;
+
+    private XContentType contentType;
+    private Map<String, Object> data;
+
+    /**
+     * Constructs a new XContentSource out of the given bytes reference.
+     */
+    public XContentSource(BytesReference bytes) throws ElasticsearchParseException {
+        this.bytes = bytes;
+    }
+
+    /**
+     * @return The bytes reference of the source
+     */
+    public BytesReference getBytes() {
+        return bytes;
+    }
+
+    /**
+     * @return The source as a map
+     */
+    public Map<String, Object> getAsMap() {
+        if (data == null) {
+            Tuple<XContentType, Map<String, Object>> tuple = XContentHelper.convertToMap(bytes, false);
+            this.contentType = tuple.v1();
+            this.data = tuple.v2();
+        }
+        return data;
+    }
+
+    /**
+     * Extracts a value identified by the given path in the source.
+     *
+     * @param path a dot notation path to the requested value
+     * @return The extracted value or {@code null} if no value is associated with the given path
+     */
+    public <T> T getValue(String path) {
+        return (T) XContentMapValues.extractValue(path, getAsMap());
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        XContentParser parser = contentType().xContent().createParser(bytes);
+        parser.nextToken();
+        XContentHelper.copyCurrentStructure(builder.generator(), parser);
+        return builder;
+    }
+
+    public static XContentSource readFrom(StreamInput in) throws IOException {
+        return new XContentSource(in.readBytesReference());
+    }
+
+    public static void writeTo(XContentSource source, StreamOutput out) throws IOException {
+        out.writeBytesReference(source.bytes);
+    }
+
+    private XContentType contentType() {
+        if (contentType == null) {
+            contentType = XContentFactory.xContentType(bytes);
+        }
+        return contentType;
+    }
+
+}

src/main/java/org/elasticsearch/watcher/transport/actions/execute/ExecuteWatchResponse.java

@@ -10,61 +10,43 @@ import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.watcher.support.xcontent.XContentSource;
 
 import java.io.IOException;
-import java.util.Map;
 
 /**
  * This class contains the WatchHistory generated by running the watch
  */
 public class ExecuteWatchResponse extends ActionResponse {
 
-    private BytesReference watchRecordSource;
+    private XContentSource source;
-
-    private Map<String, Object> watchRecordAsMap;
 
     public ExecuteWatchResponse() {
     }
 
-    public ExecuteWatchResponse(@Nullable BytesReference watchRecordSource) {
+    public ExecuteWatchResponse(@Nullable BytesReference source) {
-        this.watchRecordSource = watchRecordSource;
+        this.source = source != null ? new XContentSource(source) : null;
     }
 
     /**
-     * @return The watch record bytes reference
+     * @return The watch record source
      */
-    public BytesReference getWatchRecordSource() {
+    public XContentSource getSource() {
-        return watchRecordSource;
+        return source;
-    }
-
-    /**
-     * @return a map representation of the watch record
-     */
-    public Map<String, Object> getWatchRecordAsMap() {
-        if (watchRecordSource == null) {
-            return null;
-        }
-        if (watchRecordAsMap != null) {
-            return watchRecordAsMap;
-        }
-
-        watchRecordAsMap = XContentHelper.convertToMap(watchRecordSource, true).v2();
-        return watchRecordAsMap;
     }
 
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        watchRecordSource = in.readBoolean() ? in.readBytesReference() : null;
+        source = in.readBoolean() ? XContentSource.readFrom(in) : null;
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeBoolean(watchRecordSource != null);
+        out.writeBoolean(source != null);
-        if (watchRecordSource != null) {
+        if (source != null) {
-            out.writeBytesReference(watchRecordSource);
+            XContentSource.writeTo(source, out);
         }
     }
 }

src/main/java/org/elasticsearch/watcher/transport/actions/get/GetWatchResponse.java

@@ -5,33 +5,30 @@
  */
 package org.elasticsearch.watcher.transport.actions.get;
 
-import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.search.lookup.SourceLookup;
+import org.elasticsearch.watcher.support.xcontent.XContentSource;
 
 import java.io.IOException;
-import java.util.Map;
 
 public class GetWatchResponse extends ActionResponse {
 
     private String id;
     private long version = -1;
     private boolean found = false;
-    private BytesReference source;
+    private XContentSource source;
-
-    private Map<String, Object> sourceAsMap;
 
     GetWatchResponse() {
     }
 
     public GetWatchResponse(String id, long version, boolean found, BytesReference source) {
+        assert !found && source == null || found && source.length() > 0;
         this.id = id;
         this.version = version;
         this.found = found;
-        this.source = source;
+        this.source = found ? new XContentSource(source) : null;
     }
 
     public String getId() {
@@ -47,29 +44,17 @@ public class GetWatchResponse extends ActionResponse {
         return found;
     }
 
-    public BytesReference getSource() {
+    public XContentSource getSource() {
         return source;
     }
 
-    public Map<String, Object> getSourceAsMap() throws ElasticsearchParseException {
-        if (source == null) {
-            return null;
-        }
-        if (sourceAsMap != null) {
-            return sourceAsMap;
-        }
-
-        sourceAsMap = SourceLookup.sourceAsMap(source);
-        return sourceAsMap;
-    }
-
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         id = in.readString();
         found = in.readBoolean();
         version = in.readLong();
-        source = found ? in.readBytesReference() : null;
+        source = found ? XContentSource.readFrom(in) : null;
     }
 
     @Override
@@ -79,7 +64,7 @@ public class GetWatchResponse extends ActionResponse {
         out.writeBoolean(found);
         out.writeLong(version);
         if (found) {
-            out.writeBytesReference(source);
+            XContentSource.writeTo(source, out);
         }
     }
 }

src/test/java/org/elasticsearch/watcher/execution/ManualExecutionTests.java

@@ -116,7 +116,7 @@ public class ManualExecutionTests extends AbstractWatcherIntegrationTests {
         Watch testWatch = watchService().getWatch("_id");
         if (recordExecution) {
             refresh();
-            Watch persistedWatch = watchParser().parse("_id", true, watcherClient().getWatch(new GetWatchRequest("_id")).actionGet().getSource());
+            Watch persistedWatch = watchParser().parse("_id", true, watcherClient().getWatch(new GetWatchRequest("_id")).actionGet().getSource().getBytes());
             if (ignoreCondition || conditionAlwaysTrue) {
                 assertThat(testWatch.status().ackStatus().state(), equalTo(Watch.Status.AckStatus.State.ACKABLE));
                 assertThat(persistedWatch.status().ackStatus().state(), equalTo(Watch.Status.AckStatus.State.ACKABLE));
@@ -126,7 +126,6 @@ public class ManualExecutionTests extends AbstractWatcherIntegrationTests {
         } else {
             assertThat(parsedWatch.status().ackStatus().state(), equalTo(Watch.Status.AckStatus.State.AWAITS_EXECUTION));
         }
-
     }
 
     @Test
@@ -180,7 +179,7 @@ public class ManualExecutionTests extends AbstractWatcherIntegrationTests {
 
         Wid wid = new Wid("_watchId",1,new DateTime());
         ExecuteWatchResponse executeWatchResponse = watcherClient().prepareExecuteWatch().setId("_id").get();
-        WatchRecord watchRecord = watchRecordParser.parse(wid.value(), 1, executeWatchResponse.getWatchRecordSource());
+        WatchRecord watchRecord = watchRecordParser.parse(wid.value(), 1, executeWatchResponse.getSource().getBytes());
 
         assertThat(watchRecord.state(), equalTo(WatchRecord.State.EXECUTION_NOT_NEEDED));
         assertThat(watchRecord.execution().inputResult().payload().data().get("foo").toString(), equalTo("bar"));
@@ -194,14 +193,14 @@ public class ManualExecutionTests extends AbstractWatcherIntegrationTests {
         watcherClient().putWatch(new PutWatchRequest("_id", watchBuilder)).actionGet();
 
         executeWatchResponse = watcherClient().prepareExecuteWatch().setId("_id").setRecordExecution(true).get();
-        watchRecord = watchRecordParser.parse(wid.value(), 1, executeWatchResponse.getWatchRecordSource());
+        watchRecord = watchRecordParser.parse(wid.value(), 1, executeWatchResponse.getSource().getBytes());
 
         assertThat(watchRecord.state(), equalTo(WatchRecord.State.EXECUTED));
         assertThat(watchRecord.execution().inputResult().payload().data().get("foo").toString(), equalTo("bar"));
         assertThat(watchRecord.execution().actionsResults().get("log"), not(instanceOf(LoggingAction.Result.Simulated.class)));
 
         executeWatchResponse = watcherClient().prepareExecuteWatch().setId("_id").get();
-        watchRecord = watchRecordParser.parse(wid.value(), 1, executeWatchResponse.getWatchRecordSource());
+        watchRecord = watchRecordParser.parse(wid.value(), 1, executeWatchResponse.getSource().getBytes());
 
         assertThat(watchRecord.state(), equalTo(WatchRecord.State.THROTTLED));
     }

src/test/java/org/elasticsearch/watcher/history/WatchRecordTests.java

@@ -26,7 +26,7 @@ import org.elasticsearch.watcher.throttle.Throttler;
 import org.elasticsearch.watcher.trigger.schedule.ScheduleTriggerEvent;
 import org.elasticsearch.watcher.watch.Payload;
 import org.elasticsearch.watcher.watch.Watch;
-import org.elasticsearch.watcher.watch.WatchExecutionResult;
+import org.elasticsearch.watcher.execution.WatchExecutionResult;
 import org.junit.Test;
 
 import static org.elasticsearch.common.joda.time.DateTimeZone.UTC;

src/test/java/org/elasticsearch/watcher/license/LicenseIntegrationTests.java

@@ -11,7 +11,6 @@ import org.elasticsearch.common.inject.AbstractModule;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.inject.Module;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.xcontent.support.XContentMapValues;
 import org.elasticsearch.license.core.License;
 import org.elasticsearch.license.plugin.core.LicenseExpiredException;
 import org.elasticsearch.license.plugin.core.LicensesClientService;
@@ -25,12 +24,12 @@ import org.elasticsearch.watcher.test.AbstractWatcherIntegrationTests;
 import org.elasticsearch.watcher.transport.actions.put.PutWatchResponse;
 import org.elasticsearch.watcher.transport.actions.service.WatcherServiceResponse;
 import org.elasticsearch.watcher.watch.Watch;
+import org.elasticsearch.watcher.support.xcontent.XContentSource;
 import org.junit.Test;
 
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
-import java.util.Map;
 
 import static org.elasticsearch.index.query.FilterBuilders.termFilter;
 import static org.elasticsearch.index.query.QueryBuilders.*;
@@ -247,8 +246,8 @@ public class LicenseIntegrationTests extends AbstractWatcherIntegrationTests {
         assertBusy(new Runnable() {
             @Override
             public void run() {
-                Map<String, Object> source = watcherClient().prepareGetWatch(watchName).get().getSourceAsMap();
+                XContentSource source = watcherClient().prepareGetWatch(watchName).get().getSource();
-                assertThat(XContentMapValues.extractValue("status.ack.state", source), is((Object) "ackable"));
+                assertThat(source.getValue("status.ack.state"), is((Object) "ackable"));
             }
         });
 

src/test/java/org/elasticsearch/watcher/support/xcontent/XContentSourceTests.java

@@ -0,0 +1,40 @@
+/*
+ * 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.xcontent;
+
+import com.carrotsearch.randomizedtesting.annotations.Repeat;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.test.ElasticsearchTestCase;
+import org.junit.Test;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.*;
+import static org.hamcrest.Matchers.is;
+
+/**
+ *
+ */
+public class XContentSourceTests extends ElasticsearchTestCase {
+
+    @Test @Repeat(iterations = 20)
+    public void testToXContent() throws Exception {
+        XContentBuilder builder = randomBoolean() ? jsonBuilder() : randomBoolean() ? yamlBuilder() : smileBuilder();
+        BytesReference bytes = randomBoolean() ?
+                builder.startObject().field("key", "value").endObject().bytes() :
+                builder.startObject()
+                        .field("key_str", "value")
+                        .startArray("array_int").value(randomInt(10)).endArray()
+                        .nullField("key_null")
+                        .endObject()
+                        .bytes();
+        XContentSource source = new XContentSource(bytes);
+        XContentBuilder builder2 = XContentFactory.contentBuilder(builder.contentType());
+        BytesReference bytes2 = source.toXContent(builder2, ToXContent.EMPTY_PARAMS).bytes();
+        assertThat(bytes.array(), is(bytes2.array()));
+    }
+}

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

@@ -86,7 +86,7 @@ public class BasicWatcherTests extends AbstractWatcherIntegrationTests {
 
         GetWatchResponse getWatchResponse = watcherClient().prepareGetWatch().setId("_name").get();
         assertThat(getWatchResponse.isFound(), is(true));
-        assertThat(getWatchResponse.getSource().length(), greaterThan(0));
+        assertThat(getWatchResponse.getSource(), notNullValue());
     }
 
     @Test

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

@@ -17,9 +17,9 @@ import org.elasticsearch.watcher.support.secret.SecretService;
 import org.elasticsearch.watcher.test.AbstractWatcherIntegrationTests;
 import org.elasticsearch.watcher.transport.actions.execute.ExecuteWatchResponse;
 import org.elasticsearch.watcher.transport.actions.get.GetWatchResponse;
+import org.elasticsearch.watcher.support.xcontent.XContentSource;
 import org.elasticsearch.watcher.watch.WatchStore;
 import org.junit.After;
-import org.junit.Before;
 import org.junit.Test;
 
 import javax.mail.internet.MimeMessage;
@@ -112,8 +112,8 @@ public class EmailSecretsIntegrationTests extends AbstractWatcherIntegrationTest
         GetWatchResponse watchResponse = watcherClient.prepareGetWatch("_id").get();
         assertThat(watchResponse, notNullValue());
         assertThat(watchResponse.getId(), is("_id"));
-        source = watchResponse.getSourceAsMap();
+        XContentSource contentSource = watchResponse.getSource();
-        value = XContentMapValues.extractValue("actions._email.email.password", source);
+        value = contentSource.getValue("actions._email.email.password");
         assertThat(value, nullValue());
 
         // now we restart, to make sure the watches and their secrets are reloaded from the index properly
@@ -136,8 +136,8 @@ public class EmailSecretsIntegrationTests extends AbstractWatcherIntegrationTest
                 .setIgnoreThrottle(true)
                 .get();
         assertThat(executeResponse, notNullValue());
-        source = executeResponse.getWatchRecordAsMap();
+        contentSource = executeResponse.getSource();
-        value = XContentMapValues.extractValue("watch_execution.actions_results._email.email.success", source);
+        value = contentSource.getValue("watch_execution.actions_results._email.email.success");
         assertThat(value, notNullValue());
         assertThat(value, is((Object) Boolean.TRUE));
 

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

@@ -22,6 +22,7 @@ import org.elasticsearch.watcher.support.secret.SecretService;
 import org.elasticsearch.watcher.test.AbstractWatcherIntegrationTests;
 import org.elasticsearch.watcher.transport.actions.execute.ExecuteWatchResponse;
 import org.elasticsearch.watcher.transport.actions.get.GetWatchResponse;
+import org.elasticsearch.watcher.support.xcontent.XContentSource;
 import org.elasticsearch.watcher.watch.WatchStore;
 import org.junit.After;
 import org.junit.Before;
@@ -125,10 +126,10 @@ public class HttpSecretsIntegrationTests extends AbstractWatcherIntegrationTests
         GetWatchResponse watchResponse = watcherClient.prepareGetWatch("_id").get();
         assertThat(watchResponse, notNullValue());
         assertThat(watchResponse.getId(), is("_id"));
-        source = watchResponse.getSourceAsMap();
+        XContentSource contentSource = watchResponse.getSource();
-        value = XContentMapValues.extractValue("input.http.request.auth.basic", source);
+        value = contentSource.getValue("input.http.request.auth.basic");
         assertThat(value, notNullValue()); // making sure we have the basic auth
-        value = XContentMapValues.extractValue("input.http.request.auth.basic.password", source);
+        value = contentSource.getValue("input.http.request.auth.basic.password");
         assertThat(value, nullValue()); // and yet we don't have the password
 
         // now we restart, to make sure the watches and their secrets are reloaded from the index properly
@@ -144,8 +145,8 @@ public class HttpSecretsIntegrationTests extends AbstractWatcherIntegrationTests
                 .setIgnoreThrottle(true)
                 .get();
         assertThat(executeResponse, notNullValue());
-        source = executeResponse.getWatchRecordAsMap();
+        contentSource = executeResponse.getSource();
-        value = XContentMapValues.extractValue("watch_execution.input_result.http.http_status", source);
+        value = contentSource.getValue("watch_execution.input_result.http.http_status");
         assertThat(value, notNullValue());
         assertThat(value, is((Object) 200));
 
@@ -195,10 +196,10 @@ public class HttpSecretsIntegrationTests extends AbstractWatcherIntegrationTests
         GetWatchResponse watchResponse = watcherClient.prepareGetWatch("_id").get();
         assertThat(watchResponse, notNullValue());
         assertThat(watchResponse.getId(), is("_id"));
-        source = watchResponse.getSourceAsMap();
+        XContentSource contentSource = watchResponse.getSource();
-        value = XContentMapValues.extractValue("actions._webhook.webhook.auth.basic", source);
+        value = contentSource.getValue("actions._webhook.webhook.auth.basic");
         assertThat(value, notNullValue()); // making sure we have the basic auth
-        value = XContentMapValues.extractValue("actions._webhook.webhook.auth.basic.password", source);
+        value = contentSource.getValue("actions._webhook.webhook.auth.basic.password");
         assertThat(value, nullValue()); // and yet we don't have the password
 
         // now we restart, to make sure the watches and their secrets are reloaded from the index properly
@@ -214,13 +215,13 @@ public class HttpSecretsIntegrationTests extends AbstractWatcherIntegrationTests
                 .setIgnoreThrottle(true)
                 .get();
         assertThat(executeResponse, notNullValue());
-        source = executeResponse.getWatchRecordAsMap();
+        contentSource = executeResponse.getSource();
-        value = XContentMapValues.extractValue("watch_execution.actions_results._webhook.webhook.response.status", source);
+        value = contentSource.getValue("watch_execution.actions_results._webhook.webhook.response.status");
         assertThat(value, notNullValue());
         assertThat(value, is((Object) 200));
-        value = XContentMapValues.extractValue("watch_execution.actions_results._webhook.webhook.request.auth.username", source);
+        value = contentSource.getValue("watch_execution.actions_results._webhook.webhook.request.auth.username");
         assertThat(value, notNullValue()); // the auth username exists
-        value = XContentMapValues.extractValue("watch_execution.actions_results._webhook.webhook.request.auth.password", source);
+        value = contentSource.getValue("watch_execution.actions_results._webhook.webhook.request.auth.password");
         assertThat(value, nullValue()); // but the auth password was filtered out
 
         RecordedRequest request = webServer.takeRequest();

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

@@ -13,6 +13,7 @@ import org.elasticsearch.watcher.transport.actions.delete.DeleteWatchResponse;
 import org.elasticsearch.watcher.transport.actions.get.GetWatchRequest;
 import org.elasticsearch.watcher.transport.actions.get.GetWatchResponse;
 import org.elasticsearch.watcher.transport.actions.put.PutWatchResponse;
+import org.elasticsearch.watcher.support.xcontent.XContentSource;
 import org.junit.Test;
 
 import java.util.Map;
@@ -82,7 +83,7 @@ public class WatchCrudTests extends AbstractWatcherIntegrationTests {
         assertThat(getResponse.isFound(), is(true));
         assertThat(getResponse.getId(), is("_name"));
         assertThat(getResponse.getVersion(), is(putResponse.getVersion()));
-        Map<String, Object> source = getResponse.getSourceAsMap();
+        Map<String, Object> source = getResponse.getSource().getAsMap();
         assertThat(source, notNullValue());
         assertThat(source, hasKey("trigger"));
         assertThat(source, hasKey("input"));
@@ -101,7 +102,7 @@ public class WatchCrudTests extends AbstractWatcherIntegrationTests {
         assertThat(getResponse.getVersion(), is(-1L));
         assertThat(getResponse.isFound(), is(false));
         assertThat(getResponse.getSource(), nullValue());
-        Map<String, Object> source = getResponse.getSourceAsMap();
+        XContentSource source = getResponse.getSource();
         assertThat(source, nullValue());
     }
 

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

@@ -92,7 +92,7 @@ public class WatchMetadataTests extends AbstractWatcherIntegrationTests {
         WatchRecord.Parser parser = getInstanceFromMaster(WatchRecord.Parser.class);
         ExecuteWatchResponse executeWatchResponse = watcherClient().prepareExecuteWatch("_name").addSimulatedActions("_all").get();
 
-        WatchRecord record = parser.parse("test_run", 1, executeWatchResponse.getWatchRecordSource());
+        WatchRecord record = parser.parse("test_run", 1, executeWatchResponse.getSource().getBytes());
         assertThat(record.metadata().get("foo").toString(), equalTo("bar"));
         assertThat(record.execution().actionsResults().get("testLogger").action(), instanceOf(LoggingAction.Result.Simulated.class));
         LoggingAction.Result.Simulated simulatedResult = (LoggingAction.Result.Simulated) (record.execution().actionsResults().get("testLogger").action());

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

@@ -101,8 +101,7 @@ public class WatchThrottleTests extends AbstractWatcherIntegrationTests {
         GetWatchResponse getWatchResponse = watcherClient.prepareGetWatch("_name").get();
         assertThat(getWatchResponse.isFound(), is(true));
 
-        Watch parsedWatch = watchParser().parse(getWatchResponse.getId(), true,
+        Watch parsedWatch = watchParser().parse(getWatchResponse.getId(), true, getWatchResponse.getSource().getBytes());
-                getWatchResponse.getSource());
         assertThat(parsedWatch.status().ackStatus().state(), is(Watch.Status.AckStatus.State.AWAITS_EXECUTION));
 
         long throttledCount = docCount(HistoryStore.INDEX_PREFIX + "*", null,
@@ -245,7 +244,7 @@ public class WatchThrottleTests extends AbstractWatcherIntegrationTests {
         }
 
         GetWatchResponse watchResponse = watcherClient.getWatch(new GetWatchRequest("_name")).actionGet();
-        Watch watch = watchParser().parse("_name", true, watchResponse.getSource());
+        Watch watch = watchParser().parse("_name", true, watchResponse.getSource().getBytes());
         assertThat(watch.status().ackStatus().state(), Matchers.equalTo(Watch.Status.AckStatus.State.ACKED));
 
         refresh();