/*
 * Copyright 2017-2018 B2i Healthcare Pte Ltd, http://b2i.sg
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.b2international.index.es;

import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Map;

import org.elasticsearch.client.Client;

import com.b2international.index.IndexClient;
import com.b2international.index.IndexClientFactory;
import com.b2international.index.es.admin.EsIndexAdmin;
import com.b2international.index.es.client.EsClient;
import com.b2international.index.es.client.tcp.EsTcpClient;
import com.b2international.index.mapping.Mappings;
import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * @since 5.10
 */
public final class EsIndexClientFactory implements IndexClientFactory {

	private static final Path DEFAULT_PATH = Paths.get("target", "resources", "indexes");

	@Override
	public IndexClient createClient(String name, ObjectMapper mapper, Mappings mappings, Map<String, Object> settings) {
		final boolean persistent = settings.containsKey(DATA_DIRECTORY);
		final Object dataSetting = settings.getOrDefault(DATA_DIRECTORY, DEFAULT_PATH);
		final Object configSetting = settings.getOrDefault(CONFIG_DIRECTORY, DEFAULT_PATH);
		final Path dataDirectory = dataSetting instanceof Path ? (Path) dataSetting : Paths.get((String) dataSetting);
		final Path configDirectory = configSetting instanceof Path ? (Path) configSetting : Paths.get((String) configSetting);
		final String clusterName = (String) settings.getOrDefault(CLUSTER_NAME, DEFAULT_CLUSTER_NAME);
		
		final EsClient client;
		if (settings.containsKey(CLUSTER_URL)) {
			final String clusterUrl = (String) settings.get(CLUSTER_URL);
			final Object connectTimeoutSetting = settings.getOrDefault(CONNECT_TIMEOUT, DEFAULT_CONNECT_TIMEOUT);
			final Object socketTimeoutSetting = settings.getOrDefault(SOCKET_TIMEOUT, DEFAULT_SOCKET_TIMEOUT);
			final int connectTimeout = connectTimeoutSetting instanceof Integer ? (int) connectTimeoutSetting : Integer.parseInt((String) connectTimeoutSetting);
			final int socketTimeout = socketTimeoutSetting instanceof Integer ? (int) socketTimeoutSetting : Integer.parseInt((String) socketTimeoutSetting);
			final String username = (String) settings.getOrDefault(CLUSTER_USERNAME, "");
			final String password = (String) settings.getOrDefault(CLUSTER_PASSWORD, "");
			
			final EsClientConfiguration clientConfiguration = new EsClientConfiguration(clusterName, clusterUrl, username, password, connectTimeout, socketTimeout);
			
			client = EsClient.create(clientConfiguration);
		} else {
			// Start an embedded ES node only if a cluster URL is not set
			Client esClient = EsNode.getInstance(clusterName, configDirectory, dataDirectory, persistent).client();
			// and use the local NodeClient to communicate via the embedded node
			client = new EsTcpClient(esClient);
		}
		
		return new EsIndexClient(new EsIndexAdmin(client, name, mappings, settings), mapper);
	}
}

@@ -767,21 +767,17 @@
 rch.
-client.Client
+node.Node
 ;%0A%0Ai
@@ -1920,32 +1920,66 @@
  configSetting);
+%0A%0A%09%09// generic ES cluster settings
 %0A%09%09final String 
@@ -2066,145 +2066,8 @@
 ;%0A%09%09
-%0A%09%09final EsClient client;%0A%09%09if (settings.containsKey(CLUSTER_URL)) %7B%0A%09%09%09final String clusterUrl = (String) settings.get(CLUSTER_URL);%0A%09%09%09
 fina
@@ -2156,33 +2156,32 @@
 NNECT_TIMEOUT);%0A
-%09
 %09%09final Object s
@@ -2263,25 +2263,24 @@
 TIMEOUT);%0A%09%09
-%09
 final int co
@@ -2415,25 +2415,24 @@
 Setting);%0A%09%09
-%09
 final int so
@@ -2565,17 +2565,16 @@
 tting);%0A
-%09
 %09%09final 
@@ -2647,17 +2647,16 @@
  %22%22);%0A%09%09
-%09
 final St
@@ -2727,21 +2727,19 @@
  %22%22);%0A%09%09
-%09
 %0A%09%09
-%09
 final Es
@@ -2748,44 +2748,153 @@
 ient
-Configuration clientConfiguration = 
+ client;%0A%09%09if (settings.containsKey(CLUSTER_URL)) %7B%0A%09%09%09final String clusterUrl = (String) settings.get(CLUSTER_URL);%0A%09%09%09client = EsClient.create(
 new 
@@ -2994,61 +2994,8 @@
 out)
-;%0A%09%09%09%0A%09%09%09client = EsClient.create(clientConfiguration
 );%0A%09
@@ -3076,23 +3076,17 @@
 %0A%09%09%09
-Client esClient
+Node node
  = E
@@ -3163,19 +3163,301 @@
 ent)
-.client();%0A
+;%0A%09%09%09// check sysprop to force HTTP client when still using embedded mode%0A%09%09%09if (System.getProperty(%22so.index.es.useHttp%22) != null) %7B%0A%09%09%09%09client = EsClient.create(new EsClientConfiguration(clusterName, %22http://127.0.0.1:9200%22, username, password, connectTimeout, socketTimeout));%0A%09%09%09%7D else %7B%0A%09
 %09%09%09/
@@ -3523,24 +3523,25 @@
 ded node%0A%09%09%09
+%09
 client = new
@@ -3557,18 +3557,28 @@
 ent(
-esClient);
+node.client());%0A%09%09%09%7D
 %0A%09%09%7D

support HTTP ES client in embedded mode