MultiplyByConstant extends BaseOperator { public void setMultiplier(Number multiplier) { this.multiplier = multiplier; }  void setMultiplier(Number multiplier); Number getMultiplier(); final transient DefaultInputPort<Number> input; final transient DefaultOutputPort<Long> longProduct; final transient DefaultOutputPort<Integer> integerProduct; final transient DefaultOutputPort<Double> doubleProduct; final transient DefaultOutputPort<Float> floatProduct; }
@Test public void testNodeSchemaProcessing() { MultiplyByConstant oper = new MultiplyByConstant(); SumTestSink<Object> lmultSink = new SumTestSink<Object>(); SumTestSink<Object> imultSink = new SumTestSink<Object>(); SumTestSink<Object> dmultSink = new SumTestSink<Object>(); SumTestSink<Object> fmultSink = new SumTestSink<Object>(); oper.longProduct.setSink(lmultSink); oper.integerProduct.setSink(imultSink); oper.doubleProduct.setSink(dmultSink); oper.floatProduct.setSink(fmultSink); int imby = 2; Integer mby = imby; oper.setMultiplier(mby); oper.beginWindow(0); int sum = 0; for (int i = 0; i < 100; i++) { Integer t = i; oper.input.process(t); sum += i * imby; } oper.endWindow(); Assert.assertEquals("sum was", sum, lmultSink.val.intValue()); Assert.assertEquals("sum was", sum, imultSink.val.intValue()); Assert.assertEquals("sum was", sum, dmultSink.val.intValue()); Assert.assertEquals("sum", sum, fmultSink.val.intValue()); }