package io.korobi.mongotoelastic.processor;

import com.mongodb.client.MongoCursor;
import io.korobi.mongotoelastic.exception.ImportException;
import io.korobi.mongotoelastic.logging.InjectLogger;
import io.korobi.mongotoelastic.mongo.IChannelBlacklist;
import org.apache.logging.log4j.Logger;
import org.bson.Document;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;

import javax.inject.Inject;
import java.io.IOException;
import java.util.Date;
import java.util.concurrent.atomic.AtomicInteger;

public class MongoToElasticProcessor implements IDocumentProcessor {

    private static final AtomicInteger batchNo = new AtomicInteger();

    @InjectLogger
    private Logger logger;
    private Client esClient;
    private IChannelBlacklist blacklist;

    @Inject
    public MongoToElasticProcessor(Client esClient, IChannelBlacklist blacklist) {
        this.esClient = esClient;
        this.blacklist = blacklist;
    }

    @Override
    public void run(MongoCursor<Document> documents, int bulkSize) {
        BulkRequestBuilder bulkRequest = this.esClient.prepareBulk();
        Document doc;

        synchronized(documents) {
            doc = documents.tryNext();
        }
        int i = 0;
        while(doc != null) {
            if (this.blacklist.isBlacklisted(doc.getString("network"), doc.getString("channel"))) {
                synchronized(documents) {
                    doc = documents.tryNext();
                }
                continue;
            }

            try {
                Date date = (Date) doc.get("date");
                XContentBuilder docBuilder =
                    XContentFactory.jsonBuilder()
                        .startObject()
                            .field("actor_hostname", doc.getString("actor_hostname"))
                            .field("actor_name", doc.getString("actor_name"))
                            .field("actor_prefix", doc.getString("actor_prefix"))
                            .field("channel", doc.getString("channel"))
                            // .field("channel_mode", doc.getString("channel_mode")) // TODO: Check w/ kashike
                            .field("message", doc.getString("message"))
                            .field("network", doc.getString("network"))
                            .field("recipient_hostname", doc.getString("recipient_hostname"))
                            .field("recipient_name", doc.getString("recipient_name"))
                            .field("recipient_prefix", doc.getString("recipient_prefix"))
                            .field("type", doc.getString("type"))
                            .field("mongoId", doc.get("_id").toString())
                            .field("date", date.getTime())
                        .endObject();
                bulkRequest.add(this.esClient.prepareIndex("chats", "chat").setSource(docBuilder));
                ++i;
            } catch (IOException e) {
                logger.error(e);
            }

            if(i >= bulkSize) {
                processBulkRequest(bulkRequest);
                bulkRequest = this.esClient.prepareBulk();
                i = 0;
            }

            synchronized(documents) {
                doc = documents.tryNext();
            }
        }

        processBulkRequest(bulkRequest);
    }

    private void processBulkRequest(BulkRequestBuilder bulkRequest) {
        if (bulkRequest.numberOfActions() == 0) {
            return;
        }

        int currentNo = batchNo.incrementAndGet();
        logger.info("Processing bulk request {}.", currentNo);

        BulkResponse bulkResponse = bulkRequest.execute().actionGet();
        if (bulkResponse.hasFailures()) {
            logger.error("Bulk response failure! (this is bad) Request: {}, message: {}", currentNo, bulkResponse.getItems()[0].getFailureMessage());
            throw new ImportException(bulkResponse.getItems()[0].getFailureMessage());
        } else {
            logger.info("{} docs were processed by request {}.", bulkRequest.numberOfActions(), currentNo);
        }
    }

}

@@ -782,79 +782,8 @@
  %7B%0A%0A
-    private static final AtomicInteger batchNo = new AtomicInteger();%0A%0A
     
@@ -835,16 +835,22 @@
 private 
+final 
 Client e
@@ -870,16 +870,22 @@
 private 
+final 
 IChannel
@@ -904,16 +904,57 @@
 acklist;
+%0A    private final AtomicInteger batchNo;
 %0A%0A    @I
@@ -1108,24 +1108,68 @@
  blacklist;%0A
+        this.batchNo = new AtomicInteger();%0A
     %7D%0A%0A    @
@@ -1337,16 +1337,58 @@
 ent doc;
+%0A        int i = 0;%0A        int count = 0;
 %0A%0A      
@@ -1464,35 +1464,16 @@
       %7D%0A
-        int i = 0;%0A
         
@@ -3145,16 +3145,41 @@
    ++i;%0A
+                ++count;%0A
         
@@ -3577,24 +3577,124 @@
 lkRequest);%0A
+        logger.info(%22Thread '%7B%7D' imported %7B%7D documents.%22, Thread.currentThread().getName(), count);%0A
     %7D%0A%0A    p

Print total amount of imported document per thread