EmbeddedElaticSeacrhServer.java

import org.apache.commons.collections.bag.TransformedBag;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
import org.elasticsearch.action.admin.cluster.health.ClusterIndexHealth;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.transport.LocalTransportAddress;
import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;

import static org.elasticsearch.node.NodeBuilder.nodeBuilder;

public class EmbeddedElasticsearchServer implements SearchClientService {

    public static final String DEFAULT_DATA_DIRECTORY = "target/elasticsearch-data/";
    private static final Logger LOGGER = LoggerFactory.getLogger(EmbeddedElasticsearchServer.class);
    private final String dataDirectory;
    private final Settings elasticsearchSettings;
    private final Node node;
    private final String clusterName;
    private final Client client = null;
    private TransportClient transport_client;
    private final ArrayList<Client> esClients;

    public EmbeddedElasticsearchServer(String clusterName) {
        this.clusterName = clusterName;
        this.dataDirectory = DEFAULT_DATA_DIRECTORY + this.clusterName;
        LOGGER.info("######### Starting embedded elastic search cluster, name = {} in directory = {}", clusterName, this.dataDirectory);
        /* Clean directory before starting server */
        File dataDirectory = new File(this.dataDirectory);
        if (dataDirectory.exists()) {
            EmbeddedElasticsearchServer.deleteDataDirectory(this.dataDirectory);
        }
        this.elasticsearchSettings = ImmutableSettings.settingsBuilder()
                .put("http.enabled", "false")
                .put("client.transport.sniff", false)
                .put("cluster.name", clusterName)
                .put("node.local","false")
                .put("path.data", this.dataDirectory)
                .put("index.number_of_replicas", 0)
                .build();
//        Settings settings = ImmutableSettings.settingsBuilder().put("path.data", this.dataDirectory).put("http.enabled", "false")
//                .put("cluster.name", clusterName).build();
       // node = NodeBuilder.nodeBuilder().client(false).local(false).clusterName(clusterName)
            //    .node();
        node = createLocalNode();
       // client =  node.client();
        transport_client = new TransportClient(this.elasticsearchSettings).addTransportAddress(new
                InetSocketTransportAddress("127.0.0.1", 9300));
        esClients = new ArrayList<>();
//        client = (Client)transport_client;
        esClients.add(transport_client);

        ClusterHealthResponse healths = transport_client.admin().cluster().prepareHealth().get();
        String cluster_Name = healths.getClusterName();
        int numberOfDataNodes = healths.getNumberOfDataNodes();
        int numberOfNodes = healths.getNumberOfNodes();

        System.out.println("clusterName = " + cluster_Name);
        System.out.println("numberOfDataNodes = " + numberOfDataNodes);
        System.out.println("numberOfNodes = " + numberOfNodes);
        for (ClusterIndexHealth health : healths) {
            String index = health.getIndex();
            int numberOfShards = health.getNumberOfShards();
            int numberOfReplicas = health.getNumberOfReplicas();
            ClusterHealthStatus status = health.getStatus();

            System.out.println("\tindex = " + index);
            System.out.println("\t\tnumberOfShards = " + numberOfShards);
            System.out.println("\t\tnumberOfReplicas = " + numberOfReplicas);
            System.out.println("\t\tstatus = " + status);
        }
    }

    public Node createLocalNode() {
        return nodeBuilder()
                .client(false)
                .local(false)
                .settings(this.elasticsearchSettings)
                .clusterName(this.clusterName)
                .node();
    }

    public static void deleteDataDirectory(String directory) {
        LOGGER.info("Deleting directory = {}", directory);
        try {
            FileUtils.forceDelete(new File(directory));
        } catch (IOException e) {
            LOGGER.error("Unable to delete directory = {} due to exception = {}", directory, ExceptionUtils.getStackTrace(e));
        }
    }

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

    @Override
    public ArrayList<Client> getEsClients() {
        return esClients;
    }

    @Override
    public void shutDown() {
        String nodeName = node.settings().get("name");
        LOGGER.info("######### Stopping embedded elastic search client and node = " + nodeName);
        this.getEsClients().forEach(transport_client -> {transport_client.close();transport_client.threadPool().shutdown();});
        node.close();
        System.out.println("the node is "+node+ "the clients are"+this.getEsClients());
        LOGGER.info("######### Stopped embedded elastic search node = " + nodeName + ", is closed = " + node.isClosed());
        EmbeddedElasticsearchServer.deleteDataDirectory(this.dataDirectory);
    }

    public String getClusterName() {
        return clusterName;
    }

    public Settings getSettings() {
        return node.settings();
    }
}