package com.wizzardo.tools.collections.flow.flows;

import com.wizzardo.tools.collections.flow.Flow;
import com.wizzardo.tools.collections.flow.FlowProcessor;
import com.wizzardo.tools.collections.flow.Mapper;
import com.wizzardo.tools.misc.Unchecked;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class FlowAsync<A, B> extends FlowProcessor<A, B> implements Runnable {

    protected final static int NATIVE_THREADS_COUNT = Runtime.getRuntime().availableProcessors();

    protected final ExecutorService service;
    protected final BlockingQueue<B> output;
    protected final BlockingQueue<A> input;
    protected final Mapper<A, Flow<B>> mapper;
    protected final AtomicInteger counter = new AtomicInteger();
    protected final int queueLimit;
    protected boolean waiting = false;

    public FlowAsync(Mapper<A, Flow<B>> mapper) {
        this(NATIVE_THREADS_COUNT, mapper);
    }

    public FlowAsync(int queueLimit, Mapper<A, Flow<B>> mapper) {
        this(Executors.newFixedThreadPool(NATIVE_THREADS_COUNT), queueLimit, mapper);
    }

    public FlowAsync(ExecutorService service, int queueLimit, final Mapper<A, Flow<B>> mapper) {
        this.service = service;
        this.mapper = mapper;
        this.queueLimit = queueLimit;
        input = new LinkedBlockingQueue<A>(queueLimit);
        output = new LinkedBlockingQueue<B>();
    }

    @Override
    public void process(final A a) {
        while (!canAdd()) {
            processOutput();
            waitForInput();
        }

        add(a);
        service.submit(this);


//        try {
//            input.put(a);
//        } catch (InterruptedException e) {
//            throw Unchecked.rethrow(e);
//        }
//        service.submit(task);
//        if (!output.isEmpty())
//            processOutput();
    }

    @Override
    protected void onEnd() {
        processOutput();
        while (!isEnded()) {
            waitForOutput();
            processOutput();
        }
        super.onEnd();
    }

    protected void processOutput() {
        BlockingQueue<B> queue = this.output;
        FlowProcessor<B, ?> child = FlowAsync.this.child;
        while (!queue.isEmpty()) {
            B b = queue.poll();
            if (child != null)
                child.process(b);
        }
//
//        if (ended)
//            if (monitor.isEnded())
//                super.onEnd();
//            else {
//                waitForOutput();
//                processOutput();
//            }

    }

    public boolean isEnded() {
        return counter.get() == 0;
    }

    protected void add(A a) {
        counter.incrementAndGet();
        input.add(a);
    }

    public boolean canAdd() {
        return input.size() < queueLimit;
    }

    protected void waitForInput() {
        if (input.size() < queueLimit)
            return;

        if (!output.isEmpty())
            return;

        synchronized (this) {
            do {
                if (input.size() < queueLimit)
                    return;

                if (!output.isEmpty())
                    return;

                try {
                    this.wait();
                } catch (InterruptedException ignored) {
                }
            } while (true);
        }
    }

    protected void waitForOutput() {
        if (output.isEmpty())
            synchronized (this) {
                waiting = true;
                while (output.isEmpty() && counter.get() > 0) {
                    try {
                        this.wait();
                    } catch (InterruptedException ignored) {
                    }
                }
                waiting = false;
            }
    }

    protected A take() {
        try {
            return input.take();
        } catch (InterruptedException e) {
            throw Unchecked.rethrow(e);
        } finally {
            notifyIfWaiting();
        }
    }

    protected void endTask() {
        counter.decrementAndGet();
        notifyIfWaiting();
    }

    protected void notifyIfWaiting() {
        if (waiting)
            synchronized (this) {
                if (waiting)
                    this.notifyAll();
            }
    }

    public void doInAsyncTask() {
        try {
            Flow<B> flow = mapper.map(take());
            flow.then(new FlowProcessor<B, B>() {
                @Override
                public void process(B b) {
                    try {
                        output.put(b);
                        notifyIfWaiting();
                    } catch (InterruptedException e) {
                        throw Unchecked.rethrow(e);
                    }
                }

                @Override
                protected void onEnd() {
                    endTask();
                }
            }).execute();
        } catch (Exception ex) {
            endTask();
            throw Unchecked.rethrow(ex);
        }
    }

    @Override
    public void run() {
        doInAsyncTask();
    }
}

@@ -556,53 +556,8 @@
 ce;%0A
-    protected final BlockingQueue%3CB%3E output;%0A
     
@@ -758,16 +758,25 @@
 otected 
+volatile 
 boolean 
@@ -791,16 +791,104 @@
 = false;
+%0A    protected volatile boolean blocking = false;%0A    protected BlockingQueue%3CB%3E output;
 %0A%0A    pu
@@ -964,16 +964,20 @@
 DS_COUNT
+ * 2
 , mapper
@@ -1406,48 +1406,196 @@
     
-    output = new LinkedBlockingQueue%3CB%3E(
+%7D%0A%0A    @Override%0A    public %3CT extends FlowProcessor%3CB, C%3E, C%3E T then(T command) %7B%0A        output = new LinkedBlockingQueue%3CB%3E();%0A        blocking = true;%0A        return super.then(command
 );%0A 
@@ -1675,24 +1675,54 @@
 canAdd()) %7B%0A
+            if (blocking)%0A    
             
@@ -1827,249 +1827,8 @@
 s);%0A
-%0A%0A//        try %7B%0A//            input.put(a);%0A//        %7D catch (InterruptedException e) %7B%0A//            throw Unchecked.rethrow(e);%0A//        %7D%0A//        service.submit(task);%0A//        if (!output.isEmpty())%0A//            processOutput();%0A
     
@@ -2312,210 +2312,8 @@
   %7D%0A
-//%0A//        if (ended)%0A//            if (monitor.isEnded())%0A//                super.onEnd();%0A//            else %7B%0A//                waitForOutput();%0A//                processOutput();%0A//            %7D%0A%0A
     
@@ -2661,32 +2661,44 @@
 n;%0A%0A        if (
+blocking && 
 !output.isEmpty(
@@ -2864,16 +2864,28 @@
     if (
+blocking && 
 !output.
@@ -2916,32 +2916,64 @@
        return;%0A%0A
+                waiting = true;%0A
                 
@@ -3078,32 +3078,65 @@
               %7D%0A
+                waiting = false;%0A
             %7D wh
@@ -4316,32 +4316,74 @@
           try %7B%0A
+                        if (blocking)%0A    
                 
@@ -4401,24 +4401,25 @@
 put.put(b);%0A
+%0A
             

fix possible concurrent issues and add non-blocking mode