PushPipeline extends Processor { @Override public synchronized Pushable getPushableInput(int index) { if (m_inputPushables[index] == null) { m_inputPushables[index] = new PipelinePushable(index); } return m_inputPushables[index]; } PushPipeline(Processor p, ExecutorService service); int getFinished(); @Override synchronized Pushable getPushableInput(int index); @Override synchronized Pullable getPullableOutput(int index); @Override Processor duplicate(boolean with_state);  }
@Test(timeout=10000) public void pushTest1() { Stopwatch sw = new Stopwatch(); ExecutorService ex = Executors.newCachedThreadPool(); SlowPassthrough spt = new SlowPassthrough(1, 200); PushPipeline pp = new PushPipeline(spt, ex); QueueSink qs = new QueueSink(); Queue<Object> queue = qs.getQueue(); Pushable p = pp.getPushableInput(); Connector.connect(pp, qs); sw.start(); for (int i = 0; i < 5; i++) { p.push(i); } sw.stop(); assertTrue(sw.getDuration() < 200); Stopwatch.sleep(300); assertQueueSize(5, 5, queue); assertEquals(0, queue.remove()); assertEquals(1, queue.remove()); assertEquals(2, queue.remove()); assertEquals(3, queue.remove()); assertEquals(4, queue.remove()); }
@Test(timeout=10000) public void pushTest2() { Stopwatch sw = new Stopwatch(); ExecutorService ex = Executors.newCachedThreadPool(); UnpredictableSlowPassthrough spt = new UnpredictableSlowPassthrough(1, 200); PushPipeline pp = new PushPipeline(spt, ex); QueueSink qs = new QueueSink(); Queue<Object> queue = qs.getQueue(); Pushable p = pp.getPushableInput(); Connector.connect(pp, qs); sw.start(); for (int i = 0; i < 5; i++) { p.push(i); } sw.stop(); assertTrue(sw.getDuration() < 200); Stopwatch.sleep(300); assertQueueSize(5, 5, queue); assertEquals(0, queue.remove()); assertEquals(1, queue.remove()); assertEquals(2, queue.remove()); assertEquals(3, queue.remove()); assertEquals(4, queue.remove()); }
@Test(timeout=10000) public void pushTest3() { Stopwatch sw = new Stopwatch(); ExecutorService ex = Executors.newCachedThreadPool(); SlowPassthrough spt = new SlowPassthrough(2, 100); PushPipeline pp = new PushPipeline(spt, ex); QueueSink qs = new QueueSink(2); Queue<Object> queue = qs.getQueue(0); Pushable p0 = pp.getPushableInput(0); Pushable p1 = pp.getPushableInput(1); Connector.connect(pp, qs); sw.start(); for (int i = 0; i < 5; i++) { p0.push(i); p1.push(2 * i); } sw.stop(); assertTrue(sw.getDuration() < 200); Stopwatch.sleep(400); assertQueueSize(5, 5, queue); assertEquals(0, queue.remove()); assertEquals(1, queue.remove()); assertEquals(2, queue.remove()); assertEquals(3, queue.remove()); assertEquals(4, queue.remove()); }
@Test(timeout=10000) public void getThreshold() { long delay = 2000; int num = 50; Stopwatch sw = new Stopwatch(); ExecutorService ex = Executors.newCachedThreadPool(); LoopPassthrough spt = new LoopPassthrough(1, delay); PushPipeline pp = new PushPipeline(spt, ex); QueueSink qs = new QueueSink(1); Queue<Object> queue = qs.getQueue(); Pushable p = pp.getPushableInput(); Connector.connect(pp, qs); sw.start(); for (int i = 0; i < num; i++) { p.push(0); } while (queue.size() < num) { } sw.stop(); System.out.println("Expected: " + num * delay + "; acutal: " + sw.getDuration()); }