Watcher: Support arrays in http response payload (elastic/x-pack-elasticsearch#793)

The xcontent parser was only set to read all data to a map
which did not work, when the returned data was in form of an
array (for example the cat API is doing this, if the response
format is set to JSON).

relates elastic/x-pack-elasticsearch#351

Original commit: elastic/x-pack-elasticsearch@08ad457bf6

plugin/src/main/java/org/elasticsearch/xpack/watcher/input/http/ExecutableHttpInput.java

@@ -83,7 +83,13 @@ public class ExecutableHttpInput extends ExecutableInput<HttpInput, HttpInput.Re
                 if (input.getExtractKeys() != null) {
                     payloadMap.putAll(XContentFilterKeysUtils.filterMapOrdered(input.getExtractKeys(), parser));
                 } else {
-                    payloadMap.putAll(parser.mapOrdered());
+                    // special handling if a list is returned, i.e. JSON like [ {},{} ]
+                    XContentParser.Token token = parser.nextToken();
+                    if (token == XContentParser.Token.START_ARRAY) {
+                        payloadMap.put("data", parser.listOrderedMap());
+                    } else {
+                        payloadMap.putAll(parser.mapOrdered());
+                    }
                 }
             } catch (Exception e) {
                 throw new ElasticsearchParseException("could not parse response body [{}] it does not appear to be [{}]", type(), ctx.id(),

plugin/src/test/java/org/elasticsearch/xpack/watcher/input/http/HttpInputTests.java

@@ -6,7 +6,6 @@
 package org.elasticsearch.xpack.watcher.input.http;
 
 import io.netty.handler.codec.http.HttpHeaders;
-
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.collect.MapBuilder;
 import org.elasticsearch.common.settings.Settings;
@@ -45,6 +44,7 @@ import org.junit.Before;
 
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
@@ -56,6 +56,8 @@ import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasEntry;
 import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.nullValue;
@@ -301,6 +303,29 @@ public class HttpInputTests extends ESTestCase {
         assertThat(result.payload().data().get("_status_code"), is(200));
     }
 
+    @SuppressWarnings("unchecked")
+    public void testThatArrayJsonResponseIsHandled() throws Exception {
+        Map<String, String[]> headers = Collections.singletonMap("Content-Type", new String[]{"application/json"});
+        HttpResponse response = new HttpResponse(200, "[ { \"foo\":  \"first\" }, {  \"foo\":  \"second\"}]", headers);
+        when(httpClient.execute(any(HttpRequest.class))).thenReturn(response);
+
+        HttpRequestTemplate.Builder request = HttpRequestTemplate.builder("localhost", 8080);
+        HttpInput httpInput = InputBuilders.httpInput(request.build()).build();
+        ExecutableHttpInput input = new ExecutableHttpInput(httpInput, logger, httpClient, templateEngine);
+
+        WatchExecutionContext ctx = createWatchExecutionContext();
+        HttpInput.Result result = input.execute(ctx, new Payload.Simple());
+        assertThat(result.statusCode, is(200));
+        assertThat(result.payload().data(), not(hasKey("_value")));
+        assertThat(result.payload().data(), hasKey("data"));
+        assertThat(result.payload().data().get("data"), instanceOf(List.class));
+        List<Map<String, String>> data = (List<Map<String, String>>) result.payload().data().get("data");
+        assertThat(data, hasSize(2));
+        assertThat(data.get(0).get("foo"), is("first"));
+        assertThat(data.get(1).get("foo"), is("second"));
+    }
+
+
     private WatchExecutionContext createWatchExecutionContext() {
         Watch watch = new Watch("test-watch",
                 new ScheduleTrigger(new IntervalSchedule(new IntervalSchedule.Interval(1, IntervalSchedule.Interval.Unit.MINUTES))),