Slice extends AbstractSlice { @Override public void reset() { super.reset(); m_lastValues.clear(); } protected  Slice();  Slice(/*@ non_null @*/ Function func, /*@ non_null @*/ Processor proc,
      Function clean_func);  Slice(/*@ non_null @*/ Function func, /*@ non_null @*/ Processor proc); @Override Object printState(); @SuppressWarnings("unchecked") @Override Slice readState(Object o); @Override Slice duplicate(boolean with_state); @Override void reset();  }
@SuppressWarnings("unchecked") @Test public void testSlicerPush() { Slice sli = new Slice(Numbers.isEven, new Sum()); QueueSink qsink = new QueueSink(1); Connector.connect(sli, qsink); Pushable in = sli.getPushableInput(0); Map<Object,Object> map; for (int k = 0; k < 2; k++) { Queue<Object> queue = qsink.getQueue(0); in.push(1); map = (Map<Object,Object>) queue.poll(); assertEquals(1.0f, map.get(false)); in.push(1); map = (Map<Object,Object>) queue.poll(); assertEquals(2.0f, map.get(false)); in.push(6); map = (Map<Object,Object>) queue.poll(); assertEquals(2.0f, map.get(false)); assertEquals(6.0f, map.get(true)); in.push(2); map = (Map<Object,Object>) queue.poll(); assertEquals(2.0f, map.get(false)); assertEquals(8.0f, map.get(true)); in.push(3); map = (Map<Object,Object>) queue.poll(); assertEquals(5.0f, map.get(false)); assertEquals(8.0f, map.get(true)); sli.reset(); qsink.reset(); } assertEquals(0, in.getPosition()); assertEquals(sli, in.getProcessor()); }
@SuppressWarnings("unchecked") @Test public void testSlicerPull() { Slice sli = new Slice(Numbers.isEven, new Sum()); QueueSource source = new QueueSource().setEvents(1, 1, 6, 2, 3); Connector.connect(source, sli); Pullable p = sli.getPullableOutput(); Map<Object,Object> map; for (int k = 0; k < 2; k++) { assertEquals(0, sli.getActiveSliceCount()); map = (Map<Object,Object>) p.pull(); assertEquals(1.0f, map.get(false)); assertEquals(1, sli.getActiveSliceCount()); map = (Map<Object,Object>) p.pull(); assertEquals(2.0f, map.get(false)); assertEquals(1, sli.getActiveSliceCount()); map = (Map<Object,Object>) p.pull(); assertEquals(2.0f, map.get(false)); assertEquals(6.0f, map.get(true)); assertEquals(2, sli.getActiveSliceCount()); map = (Map<Object,Object>) p.pull(); assertEquals(2.0f, map.get(false)); assertEquals(8.0f, map.get(true)); assertEquals(2, sli.getActiveSliceCount()); map = (Map<Object,Object>) p.pull(); assertEquals(5.0f, map.get(false)); assertEquals(8.0f, map.get(true)); assertEquals(2, sli.getActiveSliceCount()); sli.reset(); source.reset(); } p.start(); p.stop(); p.dispose(); boolean got_exception = false; try { p.remove(); } catch (UnsupportedOperationException e) { got_exception = true; } assertTrue(got_exception); }
@SuppressWarnings("unchecked") @Test public void testSlicerClean() { Slice sli = new Slice(Numbers.isEven, new Sum(), new FunctionTree(Numbers.isGreaterThan, StreamVariable.X, new Constant(2))); QueueSource source = new QueueSource().setEvents(1, 1, 2, 2, 3); Connector.connect(source, sli); Pullable p = sli.getPullableOutput(); Map<Object,Object> map; for (int k = 0; k < 2; k++) { assertEquals(0, sli.getActiveSliceCount()); map = (Map<Object,Object>) p.pull(); assertEquals(1.0f, map.get(false)); assertEquals(1, sli.getActiveSliceCount()); map = (Map<Object,Object>) p.pull(); assertEquals(2.0f, map.get(false)); assertEquals(1, sli.getActiveSliceCount()); map = (Map<Object,Object>) p.pull(); assertEquals(2.0f, map.get(false)); assertEquals(2.0f, map.get(true)); assertEquals(2, sli.getActiveSliceCount()); map = (Map<Object,Object>) p.pull(); assertEquals(2.0f, map.get(false)); assertEquals(4.0f, map.get(true)); assertEquals(1, sli.getActiveSliceCount()); map = (Map<Object,Object>) p.pull(); assertEquals(5.0f, map.get(false)); assertEquals(4.0f, map.get(true)); assertEquals(0, sli.getActiveSliceCount()); sli.reset(); source.reset(); } p.start(); p.stop(); p.dispose(); boolean got_exception = false; try { p.remove(); } catch (UnsupportedOperationException e) { got_exception = true; } assertTrue(got_exception); }