Division extends BaseOperator { @Override public void endWindow() { numer.clear(); denom.clear(); index = 0; }  void emit(Number numer, Number denom); @Override void endWindow(); final transient DefaultInputPort<Number> numerator; final transient DefaultInputPort<Number> denominator; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<Long> longQuotient; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<Integer> integerQuotient; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<Double> doubleQuotient; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<Float> floatQuotient; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<Long> longRemainder; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<Integer> integerRemainder; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<Double> doubleRemainder; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<Float> floatRemainder; @OutputPortFieldAnnotation(error = true)
  final transient DefaultOutputPort<String> errordata; }
@SuppressWarnings({"rawtypes", "unchecked"}) @Test public void testNodeProcessing() { Division oper = new Division(); CollectorTestSink lqSink = new CollectorTestSink(); CollectorTestSink iqSink = new CollectorTestSink(); CollectorTestSink dqSink = new CollectorTestSink(); CollectorTestSink fqSink = new CollectorTestSink(); CollectorTestSink lrSink = new CollectorTestSink(); CollectorTestSink irSink = new CollectorTestSink(); CollectorTestSink drSink = new CollectorTestSink(); CollectorTestSink frSink = new CollectorTestSink(); CollectorTestSink eSink = new CollectorTestSink(); oper.longQuotient.setSink(lqSink); oper.integerQuotient.setSink(iqSink); oper.doubleQuotient.setSink(dqSink); oper.floatQuotient.setSink(fqSink); oper.longRemainder.setSink(lrSink); oper.doubleRemainder.setSink(drSink); oper.floatRemainder.setSink(frSink); oper.integerRemainder.setSink(irSink); oper.errordata.setSink(eSink); oper.beginWindow(0); oper.denominator.process(5); oper.numerator.process(11); oper.denominator.process(0); oper.endWindow(); Assert.assertEquals("number emitted tuples", 1, lqSink.collectedTuples.size()); Assert.assertEquals("number emitted tuples", 1, iqSink.collectedTuples.size()); Assert.assertEquals("number emitted tuples", 1, dqSink.collectedTuples.size()); Assert.assertEquals("number emitted tuples", 1, fqSink.collectedTuples.size()); Assert.assertEquals("number emitted tuples", 1, lrSink.collectedTuples.size()); Assert.assertEquals("number emitted tuples", 1, irSink.collectedTuples.size()); Assert.assertEquals("number emitted tuples", 1, drSink.collectedTuples.size()); Assert.assertEquals("number emitted tuples", 1, frSink.collectedTuples.size()); Assert.assertEquals("number emitted tuples", 1, eSink.collectedTuples.size()); Assert.assertEquals("quotient is", new Long(2), lqSink.collectedTuples.get(0)); Assert.assertEquals("quotient is", 2, iqSink.collectedTuples.get(0)); Assert.assertEquals("quotient is", 2.2, dqSink.collectedTuples.get(0)); Assert.assertEquals("quotient is", new Float(2.2), fqSink.collectedTuples.get(0)); Assert.assertEquals("quotient is", new Long(1), lrSink.collectedTuples.get(0)); Assert.assertEquals("quotient is", 1, irSink.collectedTuples.get(0)); Assert.assertEquals("quotient is", 1.0, drSink.collectedTuples.get(0)); Assert.assertEquals("quotient is", new Float(1.0), frSink.collectedTuples.get(0)); }