package dk.lessismore.nojpa.flow;

import dk.lessismore.nojpa.db.methodquery.MQL;
import dk.lessismore.nojpa.reflection.db.model.ModelObjectInterface;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

public abstract class FlowVisitor<T extends ModelObjectInterface> {

    protected final Logger log = LoggerFactory.getLogger(getClass());


    private ThreadPoolTaskExecutor executor = null;
    int currentCount = 0;

    @Bean
    public ThreadPoolTaskExecutor getExecutor() {
        if(executor == null){
            executor = new ThreadPoolTaskExecutor();
            executor.setWaitForTasksToCompleteOnShutdown(true);
            executor.setThreadNamePrefix(getThreadNamePrefix());
            executor.setCorePoolSize(getNumberOfThreads());
            executor.setMaxPoolSize(getNumberOfThreads());
            executor.setQueueCapacity(Integer.MAX_VALUE);
            executor.initialize();
        }
        return executor;
    }

    protected abstract int getNumberOfThreads();

    protected abstract int getMinimumQueueSize();

    protected abstract int getSplitLimitSize();


    protected abstract String getThreadNamePrefix();

    public int getCurrentQueueSize(){
        return getExecutor().getThreadPoolExecutor().getQueue().size();
    }

    public void runFlow() {
        final int totalCount = getTotalCount();


        while (getSplitLimitSize() + currentCount < totalCount) {
            while (getCurrentQueueSize() < getMinimumQueueSize()) {
                log.debug("WorkQueue-fill-up-start: getCurrentQueueSize("+ getCurrentQueueSize() +"), getMinimumQueueSize("+ getMinimumQueueSize() +")");
                MQL.SelectQuery<T> query = query();
                if(startFromBeginning()){
                    query.limit(getSplitLimitSize());
                } else {
                    query.limit(currentCount, currentCount + getSplitLimitSize());
                }
                log.debug("WorkQueue-fill-up-end: getCurrentQueueSize("+ getCurrentQueueSize() +"), getMinimumQueueSize("+ getMinimumQueueSize() +")");
                int oldCount = currentCount;
                query.visit(new AbstractCountingVisitor<T>() {
                            @Override
                            public void process(T t) {
                                if(currentCount % (getSplitLimitSize() / 10) == 0){
                                    log.debug("WorkQueue-PROCESS: currentCount("+ currentCount +")");
                                }
                                currentCount++;
                                getExecutor().execute(() -> doWork(t));
                            }
                        });
                if (oldCount == currentCount) {
                    // didn't visit any
                    break;
                }
            }
            log.debug("WorkQueue-is-full: getCurrentQueueSize("+ getCurrentQueueSize() +"), getMinimumQueueSize("+ getMinimumQueueSize() +")");
            int beforeQSize = getCurrentQueueSize();
            int sameCount = 0;
            for (int i = 0; i < 180; i++) {
                try {
                    Thread.sleep(1_000);
                } catch (InterruptedException e) {
                }
                if(beforeQSize == getCurrentQueueSize()){
                    log.debug("WorkQueue-has-not-changed: getCurrentQueueSize("+ getCurrentQueueSize() +"), getMinimumQueueSize("+ getMinimumQueueSize() +")");
                    sameCount++;
                }
                log.debug("WorkQueue-checking: getCurrentQueueSize("+ getCurrentQueueSize() +"), getMinimumQueueSize("+ getMinimumQueueSize() +")");
                if(getCurrentQueueSize() < getMinimumQueueSize()){
                    break;
                }
                log.debug("WorkQueue-is-full: getCurrentQueueSize("+ getCurrentQueueSize() +"), getMinimumQueueSize("+ getMinimumQueueSize() +")");
            }
            if(sameCount == 180){
                log.debug("WorkQueue-has-not-changed-for-long-time: getCurrentQueueSize("+ getCurrentQueueSize() +"), getMinimumQueueSize("+ getMinimumQueueSize() +")");
                log.error("Will now try to force a shutdown... ");
                break;
            }


        }

        // closing things
        try{
            log.error("Will close... ");
            close();
        } catch (Exception e){

        } finally {
            System.exit(-1);
        }



    }

    public abstract MQL.SelectQuery<T> query();


    public abstract boolean startFromBeginning();
    public abstract void doWork(T t);
    public abstract void close();
    public abstract int getTotalCount();

}

@@ -1655,14 +1655,8 @@
 l-up
--start
 : ge
@@ -1775,265 +1775,60 @@
     
-MQL.SelectQuery%3CT%3E query = query();%0A                if(startFromBeginning())%7B%0A                    query.limit(getSplitLimitSize());%0A                %7D else %7B%0A                    query.limit(currentCount, currentCount + getSplitLimitSize());%0A                %7D
+int start = startFromBeginning() ? 0 : currentCount;
 %0A   
@@ -2042,16 +2042,60 @@
    query
+().limit(start, start + getSplitLimitSize())
 .visit(n
@@ -2148,24 +2148,16 @@
         
-        
 @Overrid
@@ -2178,24 +2178,16 @@
         
-   
-     
 public v
@@ -2229,26 +2229,19 @@
         
-      
-  
 if
+ 
 (current
@@ -2280,16 +2280,17 @@
 0) == 0)
+ 
 %7B%0A      
@@ -2307,32 +2307,24 @@
             
-       
- 
 log.debug(%22W
@@ -2355,16 +2355,17 @@
 tCount(%22
+ 
 + curren
@@ -2372,30 +2372,23 @@
 tCount +
+ 
 %22)%22);%0A
-        
         
@@ -2429,24 +2429,16 @@
         
-        
 currentC
@@ -2469,24 +2469,16 @@
         
-   
-     
 getExecu
@@ -2529,34 +2529,18 @@
         
-        %7D%0A        
+%7D%0A
         
@@ -4080,32 +4080,33 @@
          break;%0A
+%0A
             %7D%0A%0A%0A
@@ -4309,18 +4309,16 @@
      %7D%0A%0A
-%0A%0A
     %7D%0A%0A 
@@ -4361,25 +4361,24 @@
 %3E query();%0A%0A
-%0A
     public a

rollback bad flowvisitor impl with wrong visit() method