package com.hpcloud.disruptor;

import com.google.inject.Inject;
import com.google.inject.Provider;
import com.hpcloud.configuration.MonPersisterConfiguration;
import com.hpcloud.event.StringEvent;
import com.hpcloud.event.StringEventFactory;
import com.hpcloud.event.StringEventHandler;
import com.hpcloud.event.StringEventHandlerFactory;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class DisruptorProvider implements Provider<Disruptor> {

    private MonPersisterConfiguration configuration;
    private StringEventHandlerFactory stringEventHandlerFactory;
    private Disruptor instance;

    @Inject
    public DisruptorProvider(MonPersisterConfiguration configuration,
                             StringEventHandlerFactory stringEventHandlerFactory) {
        this.configuration = configuration;
        this.stringEventHandlerFactory = stringEventHandlerFactory;
    }

    public synchronized Disruptor<StringEvent> get() {
        if (instance == null) {

            Executor executor = Executors.newCachedThreadPool();
            StringEventFactory stringEventFactory = new StringEventFactory();

            int buffersize = configuration.getDisruptorConfiguration().getBufferSize();
            Disruptor<StringEvent> disruptor = new Disruptor(stringEventFactory, buffersize, executor);

            int batchSize = configuration.getVerticaOutputProcessorConfiguration().getBatchSize();
            int numOutputProcessors = configuration.getDisruptorConfiguration().getNumProcessors();
            EventHandlerGroup<StringEvent> handlerGroup = null;
            for (int i = 0; i < numOutputProcessors; ++i) {

                StringEventHandler stringEventHandler = stringEventHandlerFactory.create(i, numOutputProcessors, batchSize);

                if (handlerGroup == null) {
                    handlerGroup = disruptor.handleEventsWith(stringEventHandler);
                } else {
                    handlerGroup.then(stringEventHandler);
                }

            }
            disruptor.start();
            instance = disruptor;
        }
        return instance;
    }
}

@@ -363,20 +363,19 @@
 tor.
-dsl.Disrupto
+EventHandle
 r;%0Ai
@@ -403,33 +403,82 @@
 tor.dsl.
-EventHandlerGroup
+Disruptor;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory
 ;%0A%0Aimpor
@@ -611,24 +611,110 @@
 sruptor%3E %7B%0A%0A
+    private static Logger logger = LoggerFactory.getLogger(DisruptorProvider.class);%0A%0A
     private 
@@ -1196,39 +1196,182 @@
 ) %7B%0A
-        if (instance == null) %7B
+%0A        logger.debug(%22Requesting instance of disruptor%22);%0A%0A        if (instance == null) %7B%0A%0A            logger.debug(%22Instance of disruptor is null. Creating disruptor...%22);
 %0A%0A  
@@ -1534,17 +1534,17 @@
 t buffer
-s
+S
 ize = co
@@ -1592,32 +1592,121 @@
 etBufferSize();%0A
+            logger.debug(%22Buffer size for instance of disruptor %5B%22 + bufferSize + %22%5D%22);%0A%0A
             Disr
@@ -1776,17 +1776,17 @@
 , buffer
-s
+S
 ize, exe
@@ -1885,32 +1885,119 @@
 getBatchSize();%0A
+            logger.debug(%22Batch size for each output processor %5B%22 + batchSize + %22%5D%22);%0A%0A
             int 
@@ -2076,24 +2076,112 @@
 ocessors();%0A
+            logger.debug(%22Number of output processors %5B%22 + numOutputProcessors + %22%5D%22);%0A%0A
             
@@ -2196,15 +2196,12 @@
 dler
-Group%3CS
+%5B%5D s
 trin
@@ -2210,30 +2210,64 @@
 vent
-%3E h
+H
 andler
-Group = null;
+s = new StringEventHandler%5BnumOutputProcessors%5D;%0A
 %0A   
@@ -2332,33 +2332,33 @@
                 
-S
+s
 tringEventHandle
@@ -2358,35 +2358,20 @@
 tHandler
- stringEventHandler
+s%5Bi%5D
  = strin
@@ -2451,74 +2451,22 @@
     
-    if (handlerGroup == null) %7B%0A                    handlerGroup =
+%7D%0A%0A           
  dis
@@ -2499,35 +2499,37 @@
 ringEventHandler
+s
 );%0A
+%0A
                 
@@ -2528,80 +2528,100 @@
     
-    %7D else %7B%0A                    handlerGroup.then(stringEventHandler
+disruptor.start();%0A            logger.debug(%22Instance of disruptor successfully started%22
 );%0A
+%0A
     
@@ -2620,37 +2620,84 @@
 );%0A%0A            
-    %7D
+logger.debug(%22Instance of disruptor fully created%22);
 %0A%0A            %7D%0A
@@ -2698,21 +2698,18 @@
     
-%7D%0A           
+instance =
  dis
@@ -2710,32 +2710,24 @@
  = disruptor
-.start()
 ;%0A          
@@ -2716,36 +2716,67 @@
 ruptor;%0A        
+%7D%0A%0A
     
+    logger.debug(%22Returning 
 instance = disru
@@ -2760,33 +2760,34 @@
 urning instance 
-=
+of
  disruptor;%0A    
@@ -2780,27 +2780,20 @@
 isruptor
+%22)
 ;%0A
-        %7D
 %0A       

Make disruptor threads parallel.