package com.kodcu.service;

import com.kodcu.config.ElasticConfiguration;
import com.kodcu.config.YamlConfiguration;
import com.kodcu.util.Constants;
import org.apache.log4j.Logger;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;

import java.io.BufferedReader;
import java.io.FileReader;
import java.nio.charset.Charset;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * Created by Hakan on 5/21/2015.
 */
public class BulkService {

    private final Logger logger = Logger.getLogger(BulkService.class);
    private final YamlConfiguration config;
    private final ElasticConfiguration client;

    public BulkService(final YamlConfiguration config, final ElasticConfiguration client) {
        this.config = config;
        this.client = client;
    }

    public void startBulkOperation() {
        // TODO: Hala geliştirime açık
        String file = config.getOutFile().concat(".json");
        Pattern pattern = Pattern.compile(Constants.CREATE_ACTION);

        //Delete index if existing
        if(client.getClient().admin().indices().prepareExists(config.getDatabase()).execute().actionGet().isExists())
            deleteCurrentIndex(config.getDatabase());

        BulkProcessor bulkProcessor = BulkProcessor.builder(client.getClient(), new BulkProcessorListener())
                .setBulkActions(1000)
                .setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB))
                .build();

        try (BufferedReader br = new BufferedReader(new FileReader(file))) {
            logger.info("Transferring data began to elasticsearch.");
            String line;
            String indexName = config.getDatabase();
            String typeName = config.getCollection();
            String id = null;

            while ((line = br.readLine()) != null) {
                Matcher matcher = pattern.matcher(line);
                if (matcher.find()) {
                    id = matcher.group(1);
                } else {
                    IndexRequest indexRequest = new IndexRequest(indexName, typeName, id);
                    indexRequest.source(line.getBytes(Charset.forName("UTF-8")));
                    bulkProcessor.add(indexRequest);
                }
            }
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex.fillInStackTrace());
        } finally {
            try {
                bulkProcessor.awaitClose(60, TimeUnit.MINUTES);
            } catch (InterruptedException ex) {
                logger.error(ex.getMessage(), ex.fillInStackTrace());
            } finally {
                client.closeNode();
            }
        }
    }

    public void deleteCurrentIndex(String indexName){
        DeleteIndexResponse delete = client.getClient().admin().indices().delete(new DeleteIndexRequest(indexName)).actionGet();
        logger.info("the current index was deleted.");
    }

}

@@ -322,16 +322,106 @@
 sponse;%0A
+import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequestBuilder;%0A
 import o
@@ -508,32 +508,84 @@
 x.IndexRequest;%0A
+import org.elasticsearch.client.IndicesAdminClient;%0A
 import org.elast
@@ -1511,130 +1511,13 @@
     
-if(client.getClient().admin().indices().prepareExists(config.getDatabase()).execute().actionGet().isExists())%0A            
+this.
 dele
@@ -3043,16 +3043,17 @@
 dexName)
+ 
 %7B%0A      
@@ -3058,34 +3058,32 @@
     
-DeleteIndexResponse delete
+IndicesAdminClient admin
  = c
@@ -3121,86 +3121,397 @@
 es()
-.delete(new DeleteIndexRequest(indexName)).actionGet();%0A        logger.info(%22t
+;%0A        IndicesExistsRequestBuilder builder = admin.prepareExists(indexName);%0A        if (builder.execute().actionGet().isExists()) %7B%0A            DeleteIndexResponse delete = admin.delete(new DeleteIndexRequest(indexName)).actionGet();%0A            if (delete.isAcknowledged())%0A                logger.info(%22The current index was deleted.%22);%0A            else%0A                logger.info(%22T
 he c
@@ -3519,32 +3519,36 @@
 rrent index was 
+not 
 deleted.%22);%0A    
@@ -3539,21 +3539,31 @@
 deleted.%22);%0A
+        %7D%0A
     %7D%0A%0A%7D%0A

Refactor deleteCurrentIndex