@Override public void process(Collection<T> collection){
  Double dResult=null;
  if (doubleResult.isConnected()) {
    doubleResult.emit(dResult=aggregateDoubles(collection));
  }
  if (floatResult.isConnected()) {
    floatResult.emit(dResult == null ? (float)(aggregateDoubles(collection)) : dResult.floatValue());
  }
  Long lResult=null;
  if (longResult.isConnected()) {
    longResult.emit(lResult=aggregateLongs(collection));
  }
  if (integerResult.isConnected()) {
    integerResult.emit(lResult == null ? (int)aggregateLongs(collection) : lResult.intValue());
  }
}

@ Override public void process ( Collection < T > collection ) { Double dResult = null ; if ( doubleResult . isConnected ( ) ) { doubleResult . emit ( dResult = aggregateDoubles ( collection ) ) ; } if ( floatResult . isConnected ( ) ) { floatResult . emit ( dResult == null ? ( float ) ( aggregateDoubles ( collection ) ) : dResult . floatValue ( ) ) ; } Long lResult = null ; if ( longResult . isConnected ( ) ) { longResult . emit ( lResult = aggregateLongs ( collection ) ) ; } if ( integerResult . isConnected ( ) ) { integerResult . emit ( lResult == null ? ( int ) aggregateLongs ( collection ) : lResult . intValue ( ) ) ; } }
aggregate calculation result is only emitted on output port if it is connected .
