Watcher: Prioritize configured response content type in HttpInput (elastic/elasticsearch#2790)

When a HTTP input has a configured response content, then this should
always be treated as preferred over the content type that is returned
by the server in order to give the user the power to decide.

This also refactors the code a bit to make it more readable.

Closes elastic/elasticsearch#2211

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

elasticsearch/x-pack/watcher/src/main/java/org/elasticsearch/xpack/watcher/input/http/ExecutableHttpInput.java

@@ -11,14 +11,14 @@ import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
-import org.elasticsearch.xpack.watcher.execution.WatchExecutionContext;
-import org.elasticsearch.xpack.watcher.input.ExecutableInput;
-import org.elasticsearch.xpack.watcher.support.Variables;
-import org.elasticsearch.xpack.watcher.support.XContentFilterKeysUtils;
 import org.elasticsearch.xpack.common.http.HttpClient;
 import org.elasticsearch.xpack.common.http.HttpRequest;
 import org.elasticsearch.xpack.common.http.HttpResponse;
 import org.elasticsearch.xpack.common.text.TextTemplateEngine;
+import org.elasticsearch.xpack.watcher.execution.WatchExecutionContext;
+import org.elasticsearch.xpack.watcher.input.ExecutableInput;
+import org.elasticsearch.xpack.watcher.support.Variables;
+import org.elasticsearch.xpack.watcher.support.XContentFilterKeysUtils;
 import org.elasticsearch.xpack.watcher.watch.Payload;
 
 import java.util.HashMap;
@@ -59,42 +59,35 @@ public class ExecutableHttpInput extends ExecutableInput<HttpInput, HttpInput.Re
             return new HttpInput.Result(request, -1, payload);
         }
 
-        XContentType contentType = response.xContentType();
+        final XContentType contentType;
-        if (input.getExpectedResponseXContentType() != null) {
+        XContentType responseContentType = response.xContentType();
-            if (contentType != input.getExpectedResponseXContentType().contentType()) {
+        if (input.getExpectedResponseXContentType() == null) {
-                logger.warn("[{}] [{}] input expected content type [{}] but read [{}] from headers", type(), ctx.id(),
+            //Attempt to auto detect content type, if not set in response
-                        input.getExpectedResponseXContentType(), contentType);
+            contentType = responseContentType != null ? responseContentType : XContentFactory.xContentType(response.body());
-            }
-            if (contentType == null) {
-                contentType = input.getExpectedResponseXContentType().contentType();
-            }
         } else {
-            //Attempt to auto detect content type
+            contentType = input.getExpectedResponseXContentType().contentType();
-            if (contentType == null) {
+            if (responseContentType != contentType) {
-                contentType = XContentFactory.xContentType(response.body());
+                logger.warn("[{}] [{}] input expected content type [{}] but read [{}] from headers, using expected one", type(), ctx.id(),
-            }
+                        input.getExpectedResponseXContentType(), responseContentType);
-        }
-
-        XContentParser parser = null;
-        if (contentType != null) {
-            try {
-                parser = contentType.xContent().createParser(response.body());
-            } catch (Exception e) {
-                throw new ElasticsearchParseException("could not parse response body [{}] it does not appear to be [{}]", type(), ctx.id(),
-                        response.body().utf8ToString(), contentType.shortName());
             }
         }
 
         final Map<String, Object> payloadMap = new HashMap<>();
+        if (contentType != null) {
+            try (XContentParser parser = contentType.xContent().createParser(response.body())) {
                 if (input.getExtractKeys() != null) {
                     payloadMap.putAll(XContentFilterKeysUtils.filterMapOrdered(input.getExtractKeys(), parser));
                 } else {
-            if (parser != null) {
                     payloadMap.putAll(parser.mapOrdered());
+                }
+            } catch (Exception e) {
+                throw new ElasticsearchParseException("could not parse response body [{}] it does not appear to be [{}]", type(), ctx.id(),
+                        response.body().utf8ToString(), contentType.shortName());
+            }
         } else {
             payloadMap.put("_value", response.body().utf8ToString());
         }
-        }
+
         if (headers.size() > 0) {
             payloadMap.put("_headers", headers);
         }

elasticsearch/x-pack/watcher/src/test/java/org/elasticsearch/xpack/watcher/input/http/HttpInputTests.java

@@ -14,13 +14,6 @@ import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.test.ESTestCase;
-import org.elasticsearch.xpack.watcher.actions.ExecutableActions;
-import org.elasticsearch.xpack.watcher.condition.always.ExecutableAlwaysCondition;
-import org.elasticsearch.xpack.watcher.execution.TriggeredExecutionContext;
-import org.elasticsearch.xpack.watcher.execution.WatchExecutionContext;
-import org.elasticsearch.xpack.watcher.input.InputBuilders;
-import org.elasticsearch.xpack.watcher.input.simple.ExecutableSimpleInput;
-import org.elasticsearch.xpack.watcher.input.simple.SimpleInput;
 import org.elasticsearch.xpack.common.http.HttpClient;
 import org.elasticsearch.xpack.common.http.HttpContentType;
 import org.elasticsearch.xpack.common.http.HttpMethod;
@@ -34,6 +27,13 @@ import org.elasticsearch.xpack.common.http.auth.basic.BasicAuth;
 import org.elasticsearch.xpack.common.http.auth.basic.BasicAuthFactory;
 import org.elasticsearch.xpack.common.text.TextTemplate;
 import org.elasticsearch.xpack.common.text.TextTemplateEngine;
+import org.elasticsearch.xpack.watcher.actions.ExecutableActions;
+import org.elasticsearch.xpack.watcher.condition.always.ExecutableAlwaysCondition;
+import org.elasticsearch.xpack.watcher.execution.TriggeredExecutionContext;
+import org.elasticsearch.xpack.watcher.execution.WatchExecutionContext;
+import org.elasticsearch.xpack.watcher.input.InputBuilders;
+import org.elasticsearch.xpack.watcher.input.simple.ExecutableSimpleInput;
+import org.elasticsearch.xpack.watcher.input.simple.SimpleInput;
 import org.elasticsearch.xpack.watcher.trigger.schedule.IntervalSchedule;
 import org.elasticsearch.xpack.watcher.trigger.schedule.ScheduleTrigger;
 import org.elasticsearch.xpack.watcher.trigger.schedule.ScheduleTriggerEvent;
@@ -58,6 +58,7 @@ import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasEntry;
 import static org.hamcrest.Matchers.hasKey;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.nullValue;
 import static org.joda.time.DateTimeZone.UTC;
 import static org.mockito.Matchers.any;
@@ -270,6 +271,24 @@ public class HttpInputTests extends ESTestCase {
         assertThat(result.payload().data().get("_headers"), equalTo(expectedHeaderMap));
     }
 
+    public void testThatExpectedContentTypeOverridesReturnedContentType() throws Exception {
+        HttpRequestTemplate template = HttpRequestTemplate.builder("localhost", 9200).fromUrl("http:://127.0.0.1:12345").build();
+        HttpInput httpInput = new HttpInput(template, HttpContentType.TEXT, null);
+        ExecutableHttpInput input = new ExecutableHttpInput(httpInput, logger, httpClient, templateEngine);
+
+        Map<String, String[]> headers = new HashMap<>(1);
+        String contentType = randomFrom("application/json", "application/json; charset=UTF-8", "text/html", "application/yaml",
+                "application/smile", "application/cbor");
+        headers.put("Content-Type", new String[] { contentType });
+        String body = "{\"foo\":\"bar\"}";
+        HttpResponse httpResponse = new HttpResponse(200, body, headers);
+        when(httpClient.execute(any())).thenReturn(httpResponse);
+
+        HttpInput.Result result = input.execute(createWatchExecutionContext(), Payload.EMPTY);
+        assertThat(result.payload().data(), hasEntry("_value", body));
+        assertThat(result.payload().data(), not(hasKey("foo")));
+    }
+
     private WatchExecutionContext createWatchExecutionContext() {
         Watch watch = new Watch("test-watch",
                 new ScheduleTrigger(new IntervalSchedule(new IntervalSchedule.Interval(1, IntervalSchedule.Interval.Unit.MINUTES))),