package swoop.pipeline;

import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;

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

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.annotations.Test;

import swoop.util.ContextBasic;

public class PipelineBuilderTest {
    
    private static Logger logger = LoggerFactory.getLogger(PipelineBuilderTest.class);

    @Test
    public void usecase_notThreaded() {
        PipelineBuilder builder = new PipelineBuilder();
        ContextBasic context = new ContextBasic().register(StringBuilder.class, new StringBuilder());
        builder.context(context) //
                .handler(new Filter("time")) //
                .handler(new Downstream("auth")) //
                .handler(new Downstream("prepare")) //
                .handler(new Upstream("gzip")) //
                .handler(new Target("bob"));

        builder.buildPipeline().invokeNext();

        String string = context.get(StringBuilder.class).toString();
        assertThat(string, equalTo("<time><auth><prepare><bob><<bob>></bob></gzip></time>"));
    }

    @Test
    public void usecase_threaded() throws InterruptedException {
        ContextBasic context = new ContextBasic().register(StringBuilder.class, new StringBuilder());
        CountDownLatch latch = new CountDownLatch(1);

        PipelineBuilder builder = new PipelineBuilder();
        builder.context(context) //
                .executor(Pipelines.threadedExecutor())//
                .handler(new UpstreamLatch(latch))//
                .handler(new Filter("time")) //
                .handler(new Downstream("auth")) //
                .handler(new Downstream("prepare")) //
                .handler(new Upstream("gzip")) //
                .handler(new Target("bob"))//
                ;

        builder.buildPipeline().invokeNext();
        latch.await();

        String string = context.get(StringBuilder.class).toString();
        assertThat(string, equalTo("<time><auth><prepare><bob><<bob>></bob></gzip></time>"));
    }
    
    @Test
    public void usecase_threaded_withAsyncJob() throws InterruptedException {
        ExecutorService backgroundExecutor = Executors.newFixedThreadPool(2);
        ContextBasic context = new ContextBasic().register(StringBuilder.class, new StringBuilder());
        CountDownLatch latch = new CountDownLatch(1);

        PipelineBuilder builder = new PipelineBuilder();
        builder.context(context) //
                .executor(Pipelines.threadedExecutor())//
                .handler(new UpstreamLatch(latch))//
                .handler(new Filter("time")) //
                .handler(new Downstream("auth")) //
                .handler(new Downstream("prepare")) //
                .handler(new Upstream("gzip")) //
                .handler(new Async("bob", backgroundExecutor))//
                ;

        builder.buildPipeline().invokeNext();
        latch.await();

        String string = context.get(StringBuilder.class).toString();
        assertThat(string, equalTo("<time><auth><prepare><<bob>></gzip></time>"));
    }
    
    public static class Async implements PipelineTargetHandler {
        private String token;
        private Executor backgroundThread;
        public Async(String token, Executor backgroundThread) {
            super();
            this.token = token;
            this.backgroundThread = backgroundThread;
        }

        @Override
        public void handleTarget(final Pipeline pipeline) {
            backgroundThread.execute(new Runnable() {
                @Override
                public void run() {
                    logger.info("Async:Target [" + token + "]");
                    pipeline.get(StringBuilder.class).append("<<").append(token).append(">>");
                    // some jobs...
                    pipeline.invokeNext();
                }
            });
        }
    }

    public static class Filter implements PipelineDownstreamHandler, PipelineUpstreamHandler {
        private String token;

        public Filter(String token) {
            super();
            this.token = token;
        }

        @Override
        public void handleDownstream(Pipeline pipeline) {
            logger.info("Filter vvv{}", token);
            pipeline.get(StringBuilder.class).append("<").append(token).append(">");
            pipeline.invokeNext();
        }

        @Override
        public void handleUpstream(Pipeline pipeline) {
            logger.info("Filter ^^^{}", token);
            pipeline.get(StringBuilder.class).append("</").append(token).append(">");
            pipeline.invokeNext();
        }
    }

    public static class Target implements PipelineDownstreamHandler, PipelineUpstreamHandler, PipelineTargetHandler {
        private String token;

        public Target(String token) {
            super();
            this.token = token;
        }

        @Override
        public void handleTarget(Pipeline pipeline) {
            logger.info("Target {}", token);
            pipeline.get(StringBuilder.class).append("<<").append(token).append(">>");
            pipeline.invokeNext();
        }

        @Override
        public void handleDownstream(Pipeline pipeline) {
            logger.info("Target vvv{}", token);
            pipeline.get(StringBuilder.class).append("<").append(token).append(">");
            pipeline.invokeNext();
        }

        @Override
        public void handleUpstream(Pipeline pipeline) {
            logger.info("Target ^^^{}", token);
            pipeline.get(StringBuilder.class).append("</").append(token).append(">");
            pipeline.invokeNext();
        }
    }

    public static class Downstream implements PipelineDownstreamHandler {
        private String token;

        public Downstream(String token) {
            super();
            this.token = token;
        }

        @Override
        public void handleDownstream(Pipeline pipeline) {
            logger.info("Downstream vvv{}", token);
            pipeline.get(StringBuilder.class).append("<").append(token).append(">");
            pipeline.invokeNext();
        }
    }

    public static class Upstream implements PipelineUpstreamHandler {
        private String token;

        public Upstream(String token) {
            super();
            this.token = token;
        }

        @Override
        public void handleUpstream(Pipeline pipeline) {
            logger.info("Upstream ^^^{}", token);
            pipeline.get(StringBuilder.class).append("</").append(token).append(">");
            pipeline.invokeNext();
        }
    }

    public static class UpstreamLatch implements PipelineUpstreamHandler {
        private CountDownLatch latch;

        public UpstreamLatch(CountDownLatch latch) {
            super();
            this.latch = latch;
        }

        @Override
        public void handleUpstream(Pipeline pipeline) {
            logger.info("Upstream ^^^ latch");
            latch.countDown();
            pipeline.invokeNext();
        }
    }
}

@@ -450,20 +450,16 @@
 rTest %7B%0A
-    
 %0A    pri
@@ -1935,32 +1935,24 @@
 t(%22bob%22))//%0A
-        
         ;%0A%0A 
@@ -2185,28 +2185,24 @@
 %3E%22));%0A    %7D%0A
-    
 %0A    @Test%0A 
@@ -2983,32 +2983,24 @@
 xecutor))//%0A
-        
         ;%0A%0A 
@@ -3226,20 +3226,16 @@
 ;%0A    %7D%0A
-    
 %0A    pub
@@ -3357,32 +3357,33 @@
 ckgroundThread;%0A
+%0A
         public A

add pipeline: more suitable for a functional approach