/*
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at

  http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.  See the License for the
specific language governing permissions and limitations
under the License.
*/
package org.apache.edgent.topology;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assume.assumeTrue;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.edgent.function.BiFunction;
import org.apache.edgent.function.Function;
import org.apache.edgent.function.Functions;
import org.apache.edgent.function.ToIntFunction;
import org.apache.edgent.topology.plumbing.PlumbingStreams;
import org.apache.edgent.topology.plumbing.Valve;
import org.apache.edgent.topology.tester.Condition;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;

import com.google.gson.JsonObject;

@Ignore
public abstract class PlumbingTest extends TopologyAbstractTest {
	

	@Test
    public void testBlockingDelay() throws Exception {
		// Timing variances on shared machines can cause this test to fail
		assumeTrue(!Boolean.getBoolean("edgent.build.ci"));

        Topology topology = newTopology();
        
        TStream<String> strings = topology.strings("a", "b", "c", "d");
        
        TStream<Long> starts = strings.map(v -> System.currentTimeMillis());
        
        // delay stream
        starts = PlumbingStreams.blockingDelay(starts, 300, TimeUnit.MILLISECONDS);
        
        // calculate delay
        starts = starts.modify(v -> System.currentTimeMillis() - v);
        
        starts = starts.filter(v -> v >= 300);
        
        Condition<Long> tc = topology.getTester().tupleCount(starts, 4);
        complete(topology, tc);
        assertTrue("valid:" + tc.getResult(), tc.valid());
    }

    @Test
    public void testBlockingThrottle() throws Exception {
		// Timing variances on shared machines can cause this test to fail
    	assumeTrue(!Boolean.getBoolean("edgent.build.ci"));

        Topology topology = newTopology();
        
        TStream<String> strings = topology.strings("a", "b", "c", "d");

        TStream<Long> emittedDelays = strings.map(v -> 0L);
        
        // throttle stream
        long[] lastEmittedTimestamp = { 0 };
        emittedDelays = PlumbingStreams.blockingThrottle(emittedDelays, 300, TimeUnit.MILLISECONDS)
                .map(t -> {
                    // compute the delay since the last emitted tuple
                    long now = System.currentTimeMillis();
                    if (lastEmittedTimestamp[0] == 0)
                        lastEmittedTimestamp[0] = now;
                    t = now - lastEmittedTimestamp[0];
                    lastEmittedTimestamp[0] = now;
                    // System.out.println("### "+t);
                    return t;
                    })
                .map(t -> {
                    // simulate 200ms downstream processing delay
                    try {
                        Thread.sleep(TimeUnit.MILLISECONDS.toMillis(200));
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        throw new RuntimeException(e);
                    } return t;
                    }) ;

        // should end up with throttled delays close to 300 (not 500 like
        // a blockingDelay() under these same conditions would yield)
        emittedDelays = emittedDelays.filter(v -> v <= 320);
        
        Condition<Long> tc = topology.getTester().tupleCount(emittedDelays, 4);
        complete(topology, tc);
        assertTrue("valid:" + tc.getResult(), tc.valid());
    }

    @Test
    public void testOneShotDelay() throws Exception {

        Topology topology = newTopology();
        
        TStream<String> strings = topology.strings("a", "b", "c", "d");
        
        TStream<Long> starts = strings.map(v -> System.currentTimeMillis());
        
        // delay stream
        starts = PlumbingStreams.blockingOneShotDelay(starts, 300, TimeUnit.MILLISECONDS);
        
        // calculate display
        starts = starts.modify(v -> System.currentTimeMillis() - v);
        
        // the first tuple shouldn't satisfy the predicate
        starts = starts.filter(v -> v < 300);
        
        Condition<Long> tc = topology.getTester().tupleCount(starts, 3);
        complete(topology, tc);
        assertTrue("valid:" + tc.getResult(), tc.valid());
    }

    public static class TimeAndId {
    	private static AtomicInteger ids = new AtomicInteger();
    	long ms;
    	final int id;
    	
    	public TimeAndId() {
    		this.ms = System.currentTimeMillis();
    		this.id = ids.incrementAndGet();
    	}
    	public TimeAndId(TimeAndId tai) {
    		this.ms = System.currentTimeMillis() - tai.ms;
    		this.id = tai.id;
    	}
    	@Override
    	public String toString() {
    		return "TAI:" + id + "@" + ms;
    	}
    	
    }
    
    @Test
    public void testPressureRelieverDrop() throws Exception {

        Topology topology = newTopology();
        
        // Verify the pressureReliever drops and retains the most recent when
        // backpressure exists.
        //
        // Here, all the tuples hit the reliever at once, the downstream processing (oneShotDelay)
        // causes a backup causing the reliever's queue to become full and drop tuples.
        // The first tuple should be processed, then the last (most recent) N (N==queue depth).
        
        String[] tuples = {"A", "B", "C", "D", "E", "F", "G", "H"};
        String[] expTuples = {"A", "F", "G", "H"};  // with queue depth of 3
        TStream<String> raw = topology.strings(tuples);
        
        TStream<String> pr = PlumbingStreams.pressureReliever(raw, Functions.unpartitioned(), 3);
        
        TStream<String> pr2 = PlumbingStreams.blockingOneShotDelay(pr, 1, TimeUnit.SECONDS);
        
        Condition<Long> tcCount = topology.getTester().tupleCount(pr2, expTuples.length);
        Condition<List<String>> contents = topology.getTester().streamContents(pr2, expTuples);
        complete(topology, tcCount);
        
        assertTrue(tcCount.valid());
        assertTrue(contents.valid());
    }
    
    @Test
    public void testPressureRelieverNoDrop() throws Exception {

        Topology topology = newTopology();
        
        // Same pipeline config as testPressureRelieverDrop but the reliever queue is
        // big enough to avoid drops
        String[] tuples = {"A", "B", "C", "D", "E", "F", "G", "H"};
        TStream<String> raw = topology.strings(tuples);
        
        TStream<String> pr = PlumbingStreams.pressureReliever(raw, Functions.unpartitioned(), 100);
        
        TStream<String> pr2 = PlumbingStreams.blockingOneShotDelay(pr, 1, TimeUnit.SECONDS);
        
        Condition<Long> tcCount = topology.getTester().tupleCount(pr2, tuples.length);
        Condition<List<String>> contents = topology.getTester().streamContents(pr2, tuples);
        complete(topology, tcCount);
        
        assertTrue(tcCount.valid());
        assertTrue(contents.valid());
    }
    
    @Test
    public void testPressureRelieverContinuous() throws Exception {
		// Timing variances on shared machines can cause this test to fail
		//assumeTrue(!Boolean.getBoolean("edgent.build.ci"));

		// Try to verify more continuous reliever behavior instead of just the
		// the other pressure reliever tests where the backpressure only exists
		// at the beginning.
		//
		// Generate @ 100tps, consume @ 5tps.  
		// With reliever depth=1, roughly should process every 20th tuple, with essentially
		// no delay in the queue (certainly less than say 50% of the consumer delay, hence < 0.5 * 200ms)
		
        Topology topology = newTopology();

        TStream<TimeAndId> raw = topology.poll(() -> new TimeAndId(), 10, TimeUnit.MILLISECONDS);
        
        TStream<TimeAndId> pr = PlumbingStreams.pressureReliever(raw, Functions.unpartitioned(), 1);
        
        TStream<TimeAndId> slow = PlumbingStreams.blockingDelay(pr, 200, TimeUnit.MILLISECONDS);
        
        // calculate the delay (queue time + consumer processing)
        TStream<TimeAndId> slowM = slow.modify(v -> new TimeAndId(v));
        
        // Also process raw that should be unaffected by the slow path
        TStream<TimeAndId> processed = raw.filter(t -> true);
        
        
        Condition<Long> tcSlowMCount = topology.getTester().atLeastTupleCount(slowM, 10);
        Condition<List<TimeAndId>> tcSlowM = topology.getTester().streamContents(slowM);
        Condition<List<TimeAndId>> tcProcessed = topology.getTester().streamContents(processed);
        complete(topology, tcSlowMCount);
        
        System.out.println(String.format("testPressureRelieverContinuous() fastCnt:%d slowCnt:%d",
                tcProcessed.getResult().size(), tcSlowM.getResult().size()));
        System.out.println("slow: "+tcSlowM.getResult());

        // No lost tuples in the fast path (successive Ids, starting @ 1)
        assertEquals("fastpath tuples dropped",
                tcProcessed.getResult().size(),
                tcProcessed.getResult().get(tcProcessed.getResult().size()-1).id);

        // No dup tuples in the fast path
        Set<TimeAndId> uniqRaw = new HashSet<>(tcProcessed.getResult());
        assertEquals("fastpath tuples duplicated", tcProcessed.getResult().size(), uniqRaw.size());

        // fastpath count should be roughly 20x the slow delayed/relieved count
        assertTrue("rawCnt:"+tcProcessed.getResult().size()+" slowMCnt:"+tcSlowM.getResult().size(),
                tcProcessed.getResult().size() >= 15 * +tcSlowM.getResult().size());
        
        // slow should process roughly every 20th tuple... not successive ones
        TimeAndId prevId = null;
        for (TimeAndId id : tcSlowM.getResult()) {
            if (prevId == null) {
                // should have processed the 1st tuple
                assertEquals("slow firstId", 1, id.id);
            }
            else {
                // seems like this could be sensitive to host load
                assertTrue("slow ids prevId:"+prevId+" id:"+id,
                        id.id >= prevId.id + 15
                        && id.id <= prevId.id + 25);
            }
            prevId = id;
        }
        
        // every slow tuple should be processed near instantaneously - shouldn't wait
        // long in the queue.
        for (TimeAndId id : tcSlowM.getResult()) {
            assertTrue("slow delays prevId:"+prevId+" id:"+id,
                    id.ms <= 300);  // 200ms consumer processing + up to %50 of that waiting  
        }
    }
    
    @Test
    public void testValveState() throws Exception {
        Valve<Integer> valve = new Valve<>();
        assertTrue(valve.isOpen());
        
        valve = new Valve<>(true);
        assertTrue(valve.isOpen());
        
        valve = new Valve<>(false);
        assertFalse(valve.isOpen());
        
        valve.setOpen(true);
        assertTrue(valve.isOpen());
        
        valve.setOpen(false);
        assertFalse(valve.isOpen());
    }
    
    @Test
    public void testValveInitiallyOpen() throws Exception {
        Topology top = newTopology("testValve");

        TStream<Integer> values = top.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        
        Valve<Integer> valve = new Valve<>();
        AtomicInteger cnt = new AtomicInteger();
        TStream<Integer> filtered = values
                                    .peek(tuple -> {
                                        // reject 4,5,6
                                        int curCnt = cnt.incrementAndGet();
                                        if (curCnt > 6)
                                            valve.setOpen(true);
                                        else if (curCnt > 3)
                                            valve.setOpen(false);
                                        })
                                    .filter(valve);

        Condition<Long> count = top.getTester().tupleCount(filtered, 7);
        Condition<List<Integer>> contents = top.getTester().streamContents(filtered, 1,2,3,7,8,9,10 );
        complete(top, count);
        assertTrue(contents.getResult().toString(), contents.valid());
    }
    
    @Test
    public void testValveInitiallyClosed() throws Exception {
        Topology top = newTopology("testValve");
        
        TStream<Integer> values = top.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        
        Valve<Integer> valve = new Valve<>(false);
        
        AtomicInteger cnt = new AtomicInteger();
        TStream<Integer> filtered = values
                                    .peek(tuple -> {
                                        // reject all but 4,5,6
                                        int curCnt = cnt.incrementAndGet();
                                        if (curCnt > 6)
                                            valve.setOpen(false);
                                        else if (curCnt > 3)
                                            valve.setOpen(true);
                                        })
                                    .filter(valve);

        Condition<Long> count = top.getTester().tupleCount(filtered, 3);
        Condition<List<Integer>> contents = top.getTester().streamContents(filtered, 4,5,6 );
        complete(top, count);
        assertTrue(contents.getResult().toString(), contents.valid());
    }
    
    private Function<Integer,JsonObject> fakeAnalytic(int channel, long period, TimeUnit unit) {
      return value -> { 
        try {
          Thread.sleep(unit.toMillis(period));
          JsonObject jo = new JsonObject();
          jo.addProperty("channel", channel);
          jo.addProperty("result", value);
          return jo;
        } catch (InterruptedException e) {
          throw new RuntimeException("channel="+channel+" interrupted", e);
        }
      };
    }

    private Function<TStream<Integer>,TStream<JsonObject>> fakePipeline(int channel, long period, TimeUnit unit) {
      return stream -> stream.map(fakeAnalytic(channel, period, unit)).filter(t->true).tag("pipeline-ch"+channel);
    }
    
    @Test
    public void testConcurrentMap() throws Exception {
        Topology top = newTopology("testConcurrentMap");
        
        int ch = 0;
        List<Function<Integer,JsonObject>> mappers = new ArrayList<>();
        mappers.add(fakeAnalytic(ch++, 100, TimeUnit.MILLISECONDS));
        mappers.add(fakeAnalytic(ch++, 100, TimeUnit.MILLISECONDS));
        mappers.add(fakeAnalytic(ch++, 100, TimeUnit.MILLISECONDS));
        mappers.add(fakeAnalytic(ch++, 100, TimeUnit.MILLISECONDS));
        mappers.add(fakeAnalytic(ch++, 100, TimeUnit.MILLISECONDS));
        mappers.add(fakeAnalytic(ch++, 100, TimeUnit.MILLISECONDS));
        // a couple much faster just in case something's amiss with queues
        mappers.add(fakeAnalytic(ch++, 3, TimeUnit.MILLISECONDS));
        mappers.add(fakeAnalytic(ch++, 13, TimeUnit.MILLISECONDS));
        
        Function<List<JsonObject>,Integer> combiner = list -> {
            int sum = 0;
            int cnt = 0;
            System.out.println("combiner: "+list);
            for(JsonObject jo : list) {
              Assert.assertEquals(cnt++, jo.get("channel").getAsInt());
              sum += jo.get("result").getAsInt();
            }
            return sum;
        };

        TStream<Integer> values = top.of(1, 2, 3);
        Integer[] resultTuples = new Integer[]{
            1*mappers.size(),
            2*mappers.size(),
            3*mappers.size(),
        };
        
        TStream<Integer> result = PlumbingStreams.concurrentMap(values, mappers, combiner);
        
        Condition<Long> count = top.getTester().tupleCount(result, 3);
        Condition<List<Integer>> contents = top.getTester().streamContents(result, resultTuples );

        long begin = System.currentTimeMillis();
        complete(top, count);
        long end = System.currentTimeMillis();

        assertTrue(contents.getResult().toString(), contents.valid());
        
        long actDuration = end - begin;
        long expMinSerialDuration = resultTuples.length * mappers.size() * 100;
        long expMinDuration = resultTuples.length * 100;
        
        System.out.println(top.getName()+" expMinDuration="+expMinDuration+" actDuration="+actDuration+" expMinSerialDuration="+expMinSerialDuration);
        
        // a gross level performance check w/concurrent channels
        if (Boolean.getBoolean("edgent.build.ci"))
          System.err.println(top.getName()+" WARNING skipped performance check on 'ci' system use");
        else
          assertTrue("expMinSerialDuration="+expMinSerialDuration+" actDuration="+actDuration, 
              actDuration < 0.5 * expMinSerialDuration);
    }
    
    @Test
    public void testConcurrent() throws Exception {
        Topology top = newTopology("testConcurrent");
        
        int ch = 0;
        List<Function<TStream<Integer>,TStream<JsonObject>>> pipelines = new ArrayList<>();
        pipelines.add(fakePipeline(ch++, 100, TimeUnit.MILLISECONDS));
        pipelines.add(fakePipeline(ch++, 100, TimeUnit.MILLISECONDS));
        pipelines.add(fakePipeline(ch++, 100, TimeUnit.MILLISECONDS));
        pipelines.add(fakePipeline(ch++, 100, TimeUnit.MILLISECONDS));
        pipelines.add(fakePipeline(ch++, 100, TimeUnit.MILLISECONDS));
        
        Function<List<JsonObject>,Integer> combiner = list -> {
            int sum = 0;
            int cnt = 0;
            System.out.println("combiner: "+list);
            for(JsonObject jo : list) {
              Assert.assertEquals(cnt++, jo.get("channel").getAsInt());
              sum += jo.get("result").getAsInt();
            }
            return sum;
        };
        
        TStream<Integer> values = top.of(1, 2, 3);
        Integer[] resultTuples = new Integer[]{
            1*pipelines.size(),
            2*pipelines.size(),
            3*pipelines.size(),
        };
        
        TStream<Integer> result = PlumbingStreams.concurrent(values, pipelines, combiner).tag("result");
        
        Condition<Long> count = top.getTester().tupleCount(result, 3);
        Condition<List<Integer>> contents = top.getTester().streamContents(result, resultTuples );

        long begin = System.currentTimeMillis();
        complete(top, count);
        long end = System.currentTimeMillis();
        
        assertTrue(contents.getResult().toString(), contents.valid());
        
        long actDuration = end - begin;
        long expMinSerialDuration = resultTuples.length * pipelines.size() * 100;
        long expMinDuration = resultTuples.length * 100;
        
        System.out.println(top.getName()+" expMinDuration="+expMinDuration+" actDuration="+actDuration+" expMinSerialDuration="+expMinSerialDuration);
        
        // a gross level performance check w/concurrent channels
        if (Boolean.getBoolean("edgent.build.ci"))
          System.err.println(top.getName()+" WARNING skipped performance check on 'ci' system use");
        else
          assertTrue("expMinSerialDuration="+expMinSerialDuration+" actDuration="+actDuration, 
              actDuration < 0.5 * expMinSerialDuration);
    }

    private BiFunction<Integer,Integer,JsonObject> fakeParallelAnalytic(long period, TimeUnit unit) {
      return (value,channel) -> { 
        try {
          Thread.sleep(unit.toMillis(period));  // simulate work for this period
          JsonObject jo = new JsonObject();
          jo.addProperty("channel", channel);
          jo.addProperty("result", value);
          return jo;
        } catch (InterruptedException e) {
          throw new RuntimeException("channel="+channel+" interrupted", e);
        }
      };
    }
    
    private BiFunction<TStream<Integer>,Integer,TStream<JsonObject>> fakeParallelPipeline(long period, TimeUnit unit) {
      return (stream,channel) -> stream
          .map(value -> fakeParallelAnalytic(period, unit).apply(value,channel))
          .filter(t->true)
          .tag("pipeline-ch"+channel);
    }
    
    private Function<JsonObject,JsonObject> fakeJsonAnalytic(int channel, long period, TimeUnit unit) {
      return jo -> { 
        try {
          Thread.sleep(unit.toMillis(period));  // simulate work for this period
          return jo;
        } catch (InterruptedException e) {
          throw new RuntimeException("channel="+channel+" interrupted", e);
        }
      };
    }
    
    @SuppressWarnings("unused")
    private BiFunction<TStream<JsonObject>,Integer,TStream<JsonObject>> fakeParallelPipelineTiming(long period, TimeUnit unit) {
      return (stream,channel) -> stream
          .map(jo -> { jo.addProperty("startPipelineMsec", System.currentTimeMillis());
                       return jo; })
          .map(fakeJsonAnalytic(channel, period, unit))
          .filter(t->true)
          .map(jo -> { jo.addProperty("endPipelineMsec", System.currentTimeMillis());
                      return jo; })
          .tag("pipeline-ch"+channel);
    }
    
    @Test
    public void testParallelMap() throws Exception {
        Topology top = newTopology("testParallelMap");
        
        BiFunction<Integer,Integer,JsonObject> mapper = 
            fakeParallelAnalytic(100, TimeUnit.MILLISECONDS);
        
        int width = 5;
        ToIntFunction<Integer> splitter = tuple -> tuple % width;
        
        Integer[] resultTuples = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15};
        TStream<Integer> values = top.of(resultTuples);
        
        TStream<JsonObject> result = PlumbingStreams.parallelMap(values, width, splitter, mapper).tag("result");
        TStream<Integer> result2 = result.map(jo -> {
            int r = jo.get("result").getAsInt();
            assertEquals(splitter.applyAsInt(r), jo.get("channel").getAsInt());
            return r;
          });
        
        Condition<Long> count = top.getTester().tupleCount(result, resultTuples.length);
        Condition<List<Integer>> contents = top.getTester().contentsUnordered(result2, resultTuples);
    
        long begin = System.currentTimeMillis();
        complete(top, count);
        long end = System.currentTimeMillis();
        
        assertTrue(contents.getResult().toString(), contents.valid());
        
        long actDuration = end - begin;
        long expMinSerialDuration = resultTuples.length * 100;
        long expMinDuration = (resultTuples.length / width) * 100;
        
        System.out.println(top.getName()+" expMinDuration="+expMinDuration+" actDuration="+actDuration+" expMinSerialDuration="+expMinSerialDuration);
        
        // a gross level performance check w/parallel channels
        if (Boolean.getBoolean("edgent.build.ci"))
          System.err.println(top.getName()+" WARNING skipped performance check on 'ci' system use");
        else
          assertTrue("expMinSerialDuration="+expMinSerialDuration+" actDuration="+actDuration, 
              actDuration < 0.5 * expMinSerialDuration);
    }
    
    @Test
    public void testParallel() throws Exception {
        Topology top = newTopology("testParallel");
        
        BiFunction<TStream<Integer>,Integer,TStream<JsonObject>> pipeline = 
            fakeParallelPipeline(100, TimeUnit.MILLISECONDS);
        
        int width = 5;
        ToIntFunction<Integer> splitter = tuple -> tuple % width;
        
        Integer[] resultTuples = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15};
        TStream<Integer> values = top.of(resultTuples);
        
        TStream<JsonObject> result = PlumbingStreams.parallel(values, width, splitter, pipeline).tag("result");
        TStream<Integer> result2 = result.map(jo -> {
            int r = jo.get("result").getAsInt();
            assertEquals(splitter.applyAsInt(r), jo.get("channel").getAsInt());
            return r;
          });
        
        Condition<Long> count = top.getTester().tupleCount(result, resultTuples.length);
        Condition<List<Integer>> contents = top.getTester().contentsUnordered(result2, resultTuples);
        
        long begin = System.currentTimeMillis();
        complete(top, count);
        long end = System.currentTimeMillis();
        
        assertTrue(contents.getResult().toString(), contents.valid());
        
        long actDuration = end - begin;
        long expMinSerialDuration = resultTuples.length * 100;
        long expMinDuration = (resultTuples.length / width) * 100;
        
        System.out.println(top.getName()+" expMinDuration="+expMinDuration+" actDuration="+actDuration+" expMinSerialDuration="+expMinSerialDuration);
        
        // a gross level performance check w/parallel channels
        if (Boolean.getBoolean("edgent.build.ci"))
          System.err.println(top.getName()+" WARNING skipped performance check on 'ci' system use");
        else
          assertTrue("expMinSerialDuration="+expMinSerialDuration+" actDuration="+actDuration, 
              actDuration < 0.5 * expMinSerialDuration);
    }
    
    @Test
    public void testParallelBalanced() throws Exception {
        Topology top = newTopology("testParallelBalanced");
        
        // arrange for even channels to process ~2x as many as odd channels.
        BiFunction<TStream<Integer>,Integer,TStream<JsonObject>> pipeline =
            (stream,ch) -> {
              long delay = (ch % 2 == 0) ? 10 : 20;
              return stream.map(fakeAnalytic(ch, delay, TimeUnit.MILLISECONDS));
            };
        
        int width = 4;
        int tupCnt = 60;
        int expEvenChCnt = 2 * (tupCnt / 3);   // even:2/3rds, odd:1/3rd 
        Integer[] resultTuples = new Integer[tupCnt];
        for (int i = 0; i < tupCnt; i++)
          resultTuples[i] = i;
        AtomicInteger[] chCounts = new AtomicInteger[width];
        for (int ch = 0; ch < width; ch++)
          chCounts[ch] = new AtomicInteger();
        
        TStream<Integer> values = top.of(resultTuples);
        
        TStream<JsonObject> result = PlumbingStreams.parallelBalanced(values, width, pipeline).tag("result");
        TStream<Integer> result2 = result.map(jo -> {
            int r = jo.get("result").getAsInt();
            int ch = jo.get("channel").getAsInt();
            chCounts[ch].incrementAndGet();
            return r;
          });
        
        Condition<Long> count = top.getTester().tupleCount(result, resultTuples.length);
        Condition<List<Integer>> contents = top.getTester().contentsUnordered(result2, resultTuples);
        
        long begin = System.currentTimeMillis();
        complete(top, count);
        long end = System.currentTimeMillis();
        
        assertTrue(contents.getResult().toString(), contents.valid());
        
        long actDuration = end - begin;
        long expMinSerialDuration = resultTuples.length * 20;
        long expMinDuration = (resultTuples.length / width) * 20;
        
        System.out.println(top.getName()+" expMinDuration="+expMinDuration+" actDuration="+actDuration+" expMinSerialDuration="+expMinSerialDuration);
        System.out.println(top.getName()+" chCounts="+Arrays.asList(chCounts));
        
        // a gross level performance check w/parallel channels
        if (Boolean.getBoolean("edgent.build.ci"))
          System.err.println(top.getName()+" WARNING skipped performance check on 'ci' system use");
        else
          assertTrue("expMinSerialDuration="+expMinSerialDuration+" actDuration="+actDuration, 
              actDuration < 0.5 * expMinSerialDuration);
        
        // Verify the balancing seemed to work.
        // On loaded systems we've seen eCnt:37 oCnt:23.  Settle for expEvenCnt +- 15%
        int thresholdCnt = (int) (expEvenChCnt * 0.15);
        int evenChCnt = 0;
        for (int ch = 0; ch < width; ch++) {
          assertTrue("ch:"+ch, chCounts[ch].get() != 0);
          if (ch % 2 == 0)
            evenChCnt += chCounts[ch].get();
        }
        assertTrue(
                String.format("evenChCnt:%d expEvenChCnt:%d +-:%d", evenChCnt, expEvenChCnt, thresholdCnt),
                evenChCnt >= (expEvenChCnt - thresholdCnt)
                && evenChCnt <= (expEvenChCnt + thresholdCnt)); 
    }
    
//    @Test
//    public void testParallelTiming() throws Exception {
//        Topology top = newTopology("testParallelTiming");
//        
//        BiFunction<TStream<JsonObject>,Integer,TStream<JsonObject>> pipeline = 
//            fakeParallelPipelineTiming(100, TimeUnit.MILLISECONDS);
//        
//        int width = 5;
//        // ToIntFunction<Integer> splitter = tuple -> tuple % width;
//        ToIntFunction<JsonObject> splitter = jo -> jo.get("result").getAsInt() % width;
//        
//        Integer[] resultTuples = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15};
//        TStream<Integer> values = top.of(resultTuples);
//        
//        TStream<JsonObject> inStream = values.map(value -> {
//            JsonObject jo = new JsonObject();
//            jo.addProperty("result", value);
//            jo.addProperty("channel", splitter.applyAsInt(jo));
//            jo.addProperty("enterParallelMsec", System.currentTimeMillis());
//            return jo;
//          });
//        TStream<JsonObject> result = PlumbingStreams.parallel(inStream, width, splitter, pipeline).tag("result");
//        TStream<Integer> result2 = result.map(jo -> {
//            jo.addProperty("exitParallelMsec", System.currentTimeMillis());
//            System.out.println("ch="+jo.get("channel").getAsInt()
//                +" endPipeline-startPipeline="
//                  +(jo.get("endPipelineMsec").getAsLong()
//                    - jo.get("startPipelineMsec").getAsLong())
//                +" exitParallel-startPipeine="
//                  +(jo.get("exitParallelMsec").getAsLong()
//                      - jo.get("startPipelineMsec").getAsLong()));
//            int r = jo.get("result").getAsInt();
//            assertEquals(splitter.applyAsInt(jo), jo.get("channel").getAsInt());
//            return r;
//          });
//        
//        Condition<Long> count = top.getTester().tupleCount(result, resultTuples.length);
//        Condition<List<Integer>> contents = top.getTester().contentsUnordered(result2, resultTuples);
//        long begin = System.currentTimeMillis();
//        complete(top, count);
//        long end = System.currentTimeMillis();
//        assertTrue(contents.getResult().toString(), contents.valid());
//        
//        long actDuration = end - begin;
//        
//        long expMinSerialDuration = resultTuples.length * 100;
//        long expMinDuration = (resultTuples.length / width) * 100;
//        
//        System.out.println(top.getName()+" expMinDuration="+expMinDuration+" actDuration="+actDuration+" expMinSerialDuration="+expMinSerialDuration);
//        
//        // a gross level performance check w/parallel channels
//        assertTrue("expMinSerialDuration="+expMinSerialDuration+" actDuration="+actDuration, 
//            actDuration < 0.5 * expMinSerialDuration);
//    }

    @Test
    public void testGate() throws Exception {
        Topology topology = newTopology("testGate");

        TStream<String> raw = topology.strings("a", "b", "c", "d", "e");

        Semaphore semaphore = new Semaphore(1);
        raw = PlumbingStreams.gate(raw, semaphore);

        ArrayList<Integer> resultAvailablePermits = new ArrayList<>();
        ArrayList<Integer> arrayResult = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            arrayResult.add(0);
            arrayResult.add(1);
        }

        raw.sink(t -> {
            //Add 0 to list because semaphore.acquire() in sync has occurred
            resultAvailablePermits.add(semaphore.availablePermits());
            semaphore.release();
            //Add 1 to list because semaphore.release() has executed
            resultAvailablePermits.add(semaphore.availablePermits());
        });

        Condition<List<String>> contents = topology.getTester()
            .streamContents(raw, "a", "b", "c", "d", "e");
        complete(topology, contents);

        assertTrue("valid:" + contents.getResult(), contents.valid());
        assertTrue("valid:" + resultAvailablePermits, resultAvailablePermits.equals(arrayResult));
    }

    @Test
    public void testGateWithLocking() throws Exception {
        Topology topology = newTopology("testGateWithLocking");

        TStream<String> raw = topology.strings("a", "b", "c", "d", "e");

        Semaphore semaphore = new Semaphore(3);
        raw = PlumbingStreams.gate(raw, semaphore);

        ArrayList<Integer> resultAvailablePermits = new ArrayList<>();
        ArrayList<Integer> arrayResult = new ArrayList<>();
        arrayResult.add(2);
        arrayResult.add(1);
        arrayResult.add(0);

        raw.sink(t -> {
            //Add number of availablePermits
            resultAvailablePermits.add(semaphore.availablePermits());
        });

        Condition<List<String>> contents = topology.getTester().streamContents(raw, "a", "b", "c");
        complete(topology, contents, 1000, TimeUnit.MILLISECONDS);

        assertTrue("valid:" + contents.getResult(), contents.valid());
        assertTrue("valid:" + resultAvailablePermits, resultAvailablePermits.equals(arrayResult));
    }

}