public Map < Aggregator , AggregatorData > roll ( long newDataWindowEndTimeMillis ) { Utils . checkState ( started , "Not started" ) ; Utils . checkState ( ! stopped , "Already stopped" ) ; Map < Aggregator , AggregatorData > result = data ; Map < Aggregator , AggregatorData > newData = new ConcurrentHashMap <> ( ) ; for ( Aggregator aggregator : aggregators ) { newData . put ( aggregator , aggregator . createAggregatorData ( newDataWindowEndTimeMillis ) ) ; } data = newData ; Map < Aggregator , AggregatorData > oldData = result ; // In case of sliding window, aggregate the data windows to get the result result = aggregateDataWindows ( result ) ; if ( currentDataWindow != null ) { currentDataWindow . setDataAndClose ( oldData ) ; } DataWindow newDataWindow = createDataWindow ( newDataWindowEndTimeMillis ) ; synchronized ( dataWindowQueue ) { dataWindowQueue . add ( newDataWindow ) ; dataWindowList = new ArrayList <> ( dataWindowQueue ) ; } currentDataWindow = newDataWindow ; return result ; }
Atomically rolls the DataWindow of all aggregators associated with the AggregatorDataProvider .