package eu.luminis.performance.elasticsearch;

import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ExecutionException;

public class Elasticsearch {

    private static final Logger logger = LoggerFactory.getLogger(Elasticsearch.class);
    private static final int port = 9300;

    private final Client client;
    private final ObjectMapper objectMapper = new ObjectMapper().setSerializationInclusion(JsonInclude.Include.NON_NULL);

    private Elasticsearch(String clusterName) throws UnknownHostException {
        this.client = getClient(clusterName);
    }

    public static Elasticsearch instance(String clusterName) throws UnknownHostException {
        return new Elasticsearch(clusterName);
    }

    public Elasticsearch ifIndexExists(String indexName, Runnable runnable) {

        boolean indexExists = client
                .admin()
                .indices()
                .exists(new IndicesExistsRequest(indexName))
                .actionGet()
                .isExists();

        if (!indexExists) runnable.run();

        logger.info("Index {} exist {}", indexName, indexExists);

        return this;
    }

    public CreateIndexResponse createIndex(String indexName) {

        logger.info("Creating index {}", indexName);

        return client.admin().indices()
                .prepareCreate(indexName)
                .setSettings(Settings.settingsBuilder().put("number_of_replicas", 0).put("number_of_shards", 1).build())
                .get();
    }

    public DeleteIndexResponse deleteIndex(String indexName) throws ExecutionException, InterruptedException {
        logger.info("Deleting index {}", indexName);
        return client.admin().indices().delete(new DeleteIndexRequest(indexName)).get();
    }

    public <T> BulkResponse bulkIndex(String indexName, List<T> sources) {
        BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();

        sources
                .parallelStream()
                .map(src -> {
                    try {
                        return Optional.of(objectMapper.writeValueAsString(src));
                    } catch (JsonProcessingException e) {
                        return Optional.empty();
                    }
                })
                .filter(Optional::isPresent)
                .forEach(src -> {
                            logger.debug(src.get().toString());
                            bulkRequestBuilder
                                    .add(new IndexRequest(indexName).source(src.get()));
                        }
                );

        logger.debug("Executing bulk request with size {}", sources.size());

        return bulkRequestBuilder.get();
    }

    private Client getClient(String clusterName) throws UnknownHostException {
        final Settings settings = Settings
                .settingsBuilder()
                .put("cluster.name", clusterName).build();

        return TransportClient
                .builder()
                .settings(settings)
                .build()
                .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), port));
    }
}

@@ -1352,85 +1352,8 @@
 pper
- = new ObjectMapper().setSerializationInclusion(JsonInclude.Include.NON_NULL)
 ;%0A%0A 
@@ -1465,32 +1465,227 @@
 t(clusterName);%0A
+        this.objectMapper = new ObjectMapper()%0A                .setSerializationInclusion(JsonInclude.Include.NON_NULL)%0A                .setSerializationInclusion(JsonInclude.Include.NON_EMPTY);%0A
     %7D%0A%0A    publi
@@ -3178,22 +3178,21 @@
 eValueAs
-String
+Bytes
 (src));%0A
@@ -3420,252 +3420,97 @@
  -%3E 
-%7B%0A                            logger.debug(src.get().toString());%0A                            bulkRequestBuilder%0A                                    .add(new IndexRequest(indexName).source(src.get()));%0A                        %7D%0A                
+bulkRequestBuilder.add(new IndexRequest(indexName, indexName).source((byte%5B%5D) src.get()))
 );%0A%0A

Fix bulk request