package net.ihiroky.niotty;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author Hiroki Itoh
 */
public class ThreadPipelineElementExecutor extends TaskLoop<ThreadPipelineElementExecutor> implements PipelineElementExecutor {

    private final Lock lock_;
    private final Condition condition_;
    private final AtomicBoolean signaled_;
    private final Set<PipelineElement<?, ?>> contextSet_;
    private final ThreadPipelineElementExecutorPool pool_;

    public ThreadPipelineElementExecutor(ThreadPipelineElementExecutorPool pool) {
        lock_ = new ReentrantLock();
        condition_ = lock_.newCondition();
        signaled_ = new AtomicBoolean();
        contextSet_ = new HashSet<>();
        pool_ = pool;
    }

    @Override
    protected void onOpen() {
    }

    @Override
    protected void onClose() {
    }

    @Override
    protected void process(int timeout) throws Exception {
        lock_.lock();
        try {
            long timeoutNanos = TimeUnit.MILLISECONDS.toNanos(timeout);
            while (signaled_.getAndSet(false) || (timeoutNanos > 0 && hasNoTask())) {
                timeoutNanos = condition_.awaitNanos(timeoutNanos);
            }
        } finally {
            lock_.unlock();
        }
    }

    @Override
    protected void wakeUp() {
        lock_.lock();
        try {
            signaled_.set(true);
            condition_.signal();
        } finally {
            lock_.unlock();
        }
    }

    @Override
    public <I> void execute(final PipelineElement<I, ?> context, final I input) {
        offerTask(new Task<ThreadPipelineElementExecutor>() {
            @Override
            public int execute(ThreadPipelineElementExecutor eventLoop) throws Exception {
                context.fire(input);
                return TIMEOUT_NO_LIMIT;
            }
        });
    }

    @Override
    public <I> void execute(final PipelineElement<I, ?> context, final I input, final TransportParameter parameter) {
        offerTask(new Task<ThreadPipelineElementExecutor>() {
            @Override
            public int execute(ThreadPipelineElementExecutor eventLoop) throws Exception {
                context.fire(input, parameter);
                return TIMEOUT_NO_LIMIT;
            }
        });
    }

    @Override
    public void execute(final PipelineElement<?, ?> context, final TransportStateEvent event) {
        offerTask(new Task<ThreadPipelineElementExecutor>() {
            @Override
            public int execute(ThreadPipelineElementExecutor eventLoop) throws Exception {
                context.fire(event);
                return TIMEOUT_NO_LIMIT;
            }
        });
    }

    @Override
    public PipelineElementExecutorPool pool() {
        return pool_;
    }

    @Override
    public void close(PipelineElement<?, ?> context) {
        synchronized (pool_.assignLock()) {
            contextSet_.remove(context);
        }
        processingMemberCount_.decrementAndGet();
    }

    /**
     * Adds a specified {@code PipelineElement} to the context set.
     * This method can be called by {@link ThreadPipelineElementExecutorPool} only.
     *
     * @param context the {@code PipelineElement} to be added
     * @return true if the context set does not contains the {@code context}
     */
    boolean accept(PipelineElement<?, ?> context) {
        processingMemberCount_.incrementAndGet();
        return contextSet_.add(context);
    }

    /**
     * Returns true if the context set contains a specified {@code PipelineElement}.
     * This method can be called by {@link ThreadPipelineElementExecutorPool} only.
     *
     * @param context the {@code PipelineElement} to be checked
     * @return true if the context set contains a specified {@code PipelineElement}.
     */
    boolean contains(PipelineElement<?, ?> context) {
        return contextSet_.contains(context);
    }
}

@@ -112,58 +112,8 @@
 it;%0A
-import java.util.concurrent.atomic.AtomicBoolean;%0A
 impo
@@ -153,16 +153,16 @@
 dition;%0A
+
 import j
@@ -489,21 +489,9 @@
 ate 
-final AtomicB
+b
 oole
@@ -809,27 +809,13 @@
 _ = 
-new AtomicBoolean()
+false
 ;%0A  
@@ -1084,32 +1084,67 @@
 ;%0A        try %7B%0A
+            if (timeout %3E 0) %7B%0A    
             long
@@ -1215,15 +1215,20 @@
     
+    
 while (
+!
 sign
@@ -1236,66 +1236,16 @@
 led_
-.getAndSet(false) %7C%7C (timeoutNanos %3E 0 && hasNoTask())
 ) %7B%0A
+    
     
@@ -1308,16 +1308,167 @@
 Nanos);%0A
+                %7D%0A            %7D else if (timeout %3C 0) %7B%0A                while (!signaled_) %7B%0A                    condition_.await();%0A                %7D%0A
         
@@ -1643,18 +1643,15 @@
 led_
-.set(
+ = 
 true
-)
 ;%0A  

Fix timeout handling