package com.gentics.mesh.search.impl;

import com.gentics.elasticsearch.client.okhttp.ElasticsearchOkClient;
import com.gentics.elasticsearch.client.okhttp.RequestBuilder;

import io.reactivex.Single;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;

import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class SearchClient extends ElasticsearchOkClient<JsonObject> {

	public SearchClient(String scheme, String hostname, int port) {
		super(scheme, hostname, port);
		setConverterFunction(JsonObject::new);
	}

	/**
	 * Check if ingest processors exist on the elasticsearch nodes.
	 *
	 * @param processorName one or multiple processor names
	 * @return true if all processors where found in the list of ingest processors
	 */
	public Single<Boolean> hasIngestProcessor(String... processorName) {
		Set<String> processorNames = new HashSet<>(Arrays.asList(processorName));
		return this.nodesInfo().async()
				.filter(response -> response.containsKey("nodes"))
				.map(response -> response.getJsonObject("nodes"))
				// we need to find one node which contains
				.map(nodes ->
						nodes.stream()
							.map(entry -> nodes.getJsonObject(entry.getKey()))
							.filter(node -> node.containsKey("ingest"))
							.map(node -> node.getJsonObject("ingest"))
							.filter(ingest -> ingest.containsKey("processors"))
							.flatMap(ingest -> {
								JsonArray processors = ingest.getJsonArray("processors");
								return IntStream.range(0, processors.size())
										.mapToObj(processors::getJsonObject)
										.filter(typeObj -> typeObj.containsKey("type"))
										.map(typeObj -> typeObj.getString("type"));
							})
							.collect(Collectors.toSet())
							.containsAll(processorNames)
				).toSingle();
	}

	/**
	 * Invoke a scroll the request.
	 * 
	 * @param scrollId
	 * @param scrollTimeout
	 * @return
	 */
	public RequestBuilder<JsonObject> scroll(String scrollId, String scrollTimeout) {
		JsonObject request = new JsonObject();
		request.put("scroll", scrollTimeout);
		request.put("scroll_id", scrollId);
		return postBuilder("_search/scroll", request);
	}

}

@@ -1094,17 +1094,16 @@
 nc()%0A%09%09%09
-%09
 .filter(
@@ -1144,25 +1144,24 @@
 nodes%22))%0A%09%09%09
-%09
 .map(respons
@@ -1198,17 +1198,16 @@
 odes%22))%0A
-%09
 %09%09%09// we
@@ -1247,17 +1247,16 @@
 ains%0A%09%09%09
-%09
 .map(nod
@@ -1265,18 +1265,16 @@
  -%3E%0A%09%09%09%09
-%09
-%09
 nodes.st
@@ -1280,18 +1280,16 @@
 tream()%0A
-%09%09
 %09%09%09%09%09.ma
@@ -1333,34 +1333,32 @@
 getKey()))%0A%09%09%09%09%09
-%09
-%09
 .filter(node -%3E 
@@ -1386,26 +1386,24 @@
 est%22))%0A%09%09%09%09%09
-%09%09
 .map(node -%3E
@@ -1429,26 +1429,24 @@
 (%22ingest%22))%0A
-%09%09
 %09%09%09%09%09.filter
@@ -1495,18 +1495,16 @@
 ))%0A%09%09%09%09%09
-%09%09
 .flatMap
@@ -1516,18 +1516,16 @@
 st -%3E %7B%0A
-%09%09
 %09%09%09%09%09%09Js
@@ -1586,18 +1586,16 @@
 ;%0A%09%09%09%09%09%09
-%09%09
 return I
@@ -1627,27 +1627,24 @@
 ors.size())%0A
-%09%09%09
 %09%09%09%09%09%09%09.mapT
@@ -1682,19 +1682,16 @@
 %0A%09%09%09%09%09%09%09
-%09%09%09
 .filter(
@@ -1733,27 +1733,24 @@
 e%22))%0A%09%09%09%09%09%09%09
-%09
-%09%09
 .map(typeObj
@@ -1790,18 +1790,16 @@
 %09%09%09%09
-%09%09
 %7D)%0A%09%09%09%09%09
 %09%09.c
@@ -1794,18 +1794,16 @@
 %7D)%0A%09%09%09%09%09
-%09
-%09
 .collect
@@ -1819,26 +1819,24 @@
 rs.toSet())%0A
-%09%09
 %09%09%09%09%09.contai
@@ -1857,17 +1857,16 @@
 rNames)%0A
-%09
 %09%09%09).toS

Improve formatting