NonBlockingPush extends Processor { @Override public Pushable getPushableInput(int index) { if (this.m_inputPushables[index] == null) { m_inputPushables[index] = new NonBlockingPushable(index); } return m_inputPushables[index]; } NonBlockingPush(Processor p, ExecutorService service); @Override Pushable getPushableInput(int index); @Override Pullable getPullableOutput(int index); @Override Processor duplicate(boolean with_state);  }
@Test(timeout=10000) public void test1() throws InterruptedException { int num_instances = 100, delay = 300; ExecutorService es = Executors.newCachedThreadPool(); @SuppressWarnings("unused") List<Future<Pushable>> futures = new ArrayList<Future<Pushable>>(num_instances); List<Pushable> pushables = new ArrayList<Pushable>(num_instances); List<QueueSink> sinks = new ArrayList<QueueSink>(num_instances); for (int i = 0; i < num_instances; i++) { SlowPassthrough spt = new SlowPassthrough(1, delay); NonBlockingPush nbp = new NonBlockingPush(spt, es); QueueSink sink = new QueueSink(); Connector.connect(nbp, sink); sinks.add(sink); Pushable p = nbp.getPushableInput(); pushables.add(p); } Stopwatch sw = new Stopwatch(); sw.start(); int i = 0; for (Pushable p : pushables) { p.push(i++); } sw.stop(); assertTrue(sw.getDuration() < (num_instances * delay) * 0.2); es.awaitTermination((long) (delay * 1.2), TimeUnit.MILLISECONDS); i = 0; for (QueueSink qs : sinks) { Object o = qs.getQueue().remove(); assertEquals(i, o); i++; } }
@Test(timeout=10000) public void test2() throws InterruptedException { int num_instances = 100, delay = 300; ExecutorService es = Executors.newCachedThreadPool(); @SuppressWarnings("unused") List<Future<Pushable>> futures = new ArrayList<Future<Pushable>>(num_instances); List<Pushable> pushables0 = new ArrayList<Pushable>(num_instances); List<Pushable> pushables1 = new ArrayList<Pushable>(num_instances); List<QueueSink> sinks = new ArrayList<QueueSink>(num_instances); for (int i = 0; i < num_instances; i++) { SlowPassthrough spt = new SlowPassthrough(2, delay); NonBlockingPush nbp = new NonBlockingPush(spt, es); QueueSink sink = new QueueSink(2); Connector.connect(nbp, sink); sinks.add(sink); pushables0.add(nbp.getPushableInput(0)); pushables1.add(nbp.getPushableInput(1)); } Stopwatch sw = new Stopwatch(); sw.start(); for (int i = 0; i < num_instances; i++) { pushables0.get(i).push(i); pushables1.get(i).push(2 * i); } sw.stop(); assertTrue(sw.getDuration() < (num_instances * delay) * 0.2); es.awaitTermination((long) (delay * 1.2), TimeUnit.MILLISECONDS); for (int i = 0; i < num_instances; i++) { QueueSink qs = sinks.get(i); Object o0 = qs.getQueue(0).remove(); assertEquals(i, o0); Object o1 = qs.getQueue(1).remove(); assertEquals(2 * i, o1); } }