/*******************************************************************************
 * Copyright (c) 2011, 2017 Eurotech and/or its affiliates and others
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * Contributors:
 *     Eurotech - initial API and implementation
 *     Red Hat Inc
 *******************************************************************************/
package org.eclipse.kapua.service.datastore.internal.elasticsearch;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Map;

import org.eclipse.kapua.service.datastore.internal.setting.DatastoreSettingKey;
import org.eclipse.kapua.service.datastore.internal.setting.DatastoreSettings;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

/**
 * Elasticsearch transport client implementation.
 *
 * @since 1.0
 *
 */
public class EsTransportClientProvider implements ElasticsearchClientProvider {

    private static final int DEFAULT_PORT = 9300;

    private Client client;

    private static String[] getNodeParts(String node) {
        if (node == null)
            return new String[] {};

        String[] split = node.split(":");
        return split;
    }

    private static Client getEsClient(String hostname, int port, String clustername) throws UnknownHostException {

        Settings settings = Settings.settingsBuilder()
                .put("cluster.name", clustername).build();

        InetSocketTransportAddress ita = new InetSocketTransportAddress(InetAddress.getByName(hostname), port);
        Client esClient = TransportClient.builder().settings(settings).build().addTransportAddress(ita);

        return esClient;
    }

    /**
     * Create a new Elasticsearch transport client based on the configuration parameters ({@link DatastoreSettingKey})
     *
     * @throws EsClientUnavailableException
     */
    public EsTransportClientProvider()
            throws EsClientUnavailableException {
        DatastoreSettings config = DatastoreSettings.getInstance();
        Map<String, String> map = config.getMap(String.class, DatastoreSettingKey.ELASTICSEARCH_NODES, "[0-9]+");
        String[] esNodes = new String[] {};
        if (map != null)
            esNodes = map.values().toArray(new String[] {});

        if (esNodes == null || esNodes.length == 0)
            throw new EsClientUnavailableException("No elasticsearch nodes found");

        String[] nodeParts = getNodeParts(esNodes[0]);
        String esHost = null;
        final int esPort = config.getInt(DatastoreSettingKey.ELASTICSEARCH_PORT, DEFAULT_PORT);

        if (nodeParts.length > 0)
            esHost = nodeParts[0];

        if (nodeParts.length > 1) {
            try {
                Integer.parseInt(nodeParts[1]);
            } catch (NumberFormatException e) {
                throw new EsClientUnavailableException("Could not parse elasticsearch port: " + nodeParts[1]);
            }
        }

        Client theClient = null;
        try {
            theClient = getEsClient(esHost, esPort, config.getString(DatastoreSettingKey.ELASTICSEARCH_CLUSTER));
        } catch (UnknownHostException e) {
            throw new EsClientUnavailableException("Unknown elasticsearch node host", e);
        }

        client = theClient;

    }

    @Override
    public Client getClient() {
        return client;
    }

}

@@ -1325,16 +1325,22 @@
 private 
+final 
 Client c
@@ -1424,24 +1424,26 @@
 ode == null)
+ %7B
 %0A           
@@ -1466,16 +1466,26 @@
 ng%5B%5D %7B%7D;
+%0A        %7D
 %0A%0A      
@@ -2555,16 +2555,18 @@
 != null)
+ %7B
 %0A       
@@ -2618,16 +2618,26 @@
 g%5B%5D %7B%7D);
+%0A        %7D
 %0A%0A      
@@ -2681,16 +2681,18 @@
 th == 0)
+ %7B
 %0A       
@@ -2767,16 +2767,26 @@
 found%22);
+%0A        %7D
 %0A%0A      
@@ -2994,16 +2994,18 @@
 gth %3E 0)
+ %7B
 %0A       
@@ -3031,16 +3031,26 @@
 arts%5B0%5D;
+%0A        %7D
 %0A%0A      

Apply cleanup profile