component/src/main/java/org/wso2/extension/siddhi/execution/extrema/MaxByMinByTimeBatchWindowProcessor.java
@Override
    protected void process(ComplexEventChunk<StreamEvent> streamEventChunk, Processor nextProcessor,
                           StreamEventCloner streamEventCloner, ExtensionState state) {
        synchronized (this) {
            long currentTime = siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime();
            if (nextEmitTime == -1) {
                if (isStartTimeEnabled) {
                    nextEmitTime = getNextEmitTime(currentTime);
                } else {
                    nextEmitTime = siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime()
                            + timeInMilliSeconds;
                }
                scheduler.notifyAt(nextEmitTime);
            }

            boolean sendEvents;
            if (currentTime >= nextEmitTime) {
                nextEmitTime += timeInMilliSeconds;
                scheduler.notifyAt(nextEmitTime);
                sendEvents = true;
            } else {
                sendEvents = false;
            }

            //for each event, set the min/max event as current event
            while (streamEventChunk.hasNext()) {
                StreamEvent streamEvent = streamEventChunk.next();
                if (streamEvent.getType() != ComplexEvent.Type.CURRENT) {
                    continue;
                }
                StreamEvent clonedStreamEvent = streamEventCloner.copyStreamEvent(streamEvent);
                if (maxByMinByType.equals(MaxByMinByConstants.MIN_BY)) {
                    state.currentEvent = MaxByMinByExecutor
                            .getMinEventBatchProcessor(clonedStreamEvent, state.currentEvent, sortByAttribute);
                } else if (maxByMinByType.equals(MaxByMinByConstants.MAX_BY)) {
                    state.currentEvent = MaxByMinByExecutor
                            .getMaxEventBatchProcessor(clonedStreamEvent, state.currentEvent, sortByAttribute);
                }
            }
            streamEventChunk.clear();
            if (sendEvents) {
                if (state.expiredEventChunk.getFirst() != null) {
                    while (state.expiredEventChunk.hasNext()) {
                        StreamEvent expiredEvent = state.expiredEventChunk.next();
                        expiredEvent.setTimestamp(currentTime);
                    }
                    if (outputExpectsExpiredEvents) {
                        streamEventChunk.add(state.expiredEventChunk.getFirst());
                    }
                    state.resetEvent = streamEventCloner.copyStreamEvent(state.expiredEventChunk.getFirst());
                    // add reset event before the current events
                    if (state.resetEvent != null) {
                        state.resetEvent.setType(ComplexEvent.Type.RESET);
                        streamEventChunk.add(state.resetEvent);
                    }
                    state.resetEvent = null;
                }

                if (state.expiredEventChunk != null) {
                    state.expiredEventChunk.clear();
                }
                if (state.currentEvent != null) {
                    StreamEvent toExpireEvent = streamEventCloner.copyStreamEvent(state.currentEvent);
                    toExpireEvent.setType(StreamEvent.Type.EXPIRED);
                    state.expiredEventChunk.add(toExpireEvent);
                    streamEventChunk.add(state.currentEvent);
                }
                state.currentEvent = null;

            }
        }
        if (streamEventChunk.getFirst() != null) {
            streamEventChunk.setBatch(true);
            nextProcessor.process(streamEventChunk);
            streamEventChunk.setBatch(false);
        }
    }
/**
     * The main processing method that will be called upon event arrival
     *
     * @param streamEventChunk  the stream event chunk that need to be processed
     * @param nextProcessor     the next processor to which the success events need to be passed
     * @param streamEventCloner helps to clone the incoming event for local storage or modification
     */
The main processing method that will be called upon event arrival
@param streamEventChunk  the stream event chunk that need to be processed
@param nextProcessor     the next processor to which the success events need to be passed
@param streamEventCloner helps to clone the incoming event for local storage or modification
The main processing method that will be called upon event arrival @param streamEventChunk the stream event chunk that need to be processed @param nextProcessor the next processor to which the success events need to be passed @param streamEventCloner helps to clone the incoming event for local storage or modification
@Override
    protected void process(ComplexEventChunk<StreamEvent> streamEventChunk, Processor nextProcessor,
                           StreamEventCloner streamEventCloner, ExtensionState state) {
        synchronized (this) {
            long currentTime = siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime();
            if (nextEmitTime == -1) {
                if (isStartTimeEnabled) {
                    nextEmitTime = getNextEmitTime(currentTime);
                } else {
                    nextEmitTime = siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime()
                            + timeInMilliSeconds;
                }
                scheduler.notifyAt(nextEmitTime);
            }
            boolean sendEvents;
            if (currentTime >= nextEmitTime) {
                nextEmitTime += timeInMilliSeconds;
                scheduler.notifyAt(nextEmitTime);
                sendEvents = true;
            } else {
                sendEvents = false;
            }
            while (streamEventChunk.hasNext()) {
                StreamEvent streamEvent = streamEventChunk.next();
                if (streamEvent.getType() != ComplexEvent.Type.CURRENT) {
                    continue;
                }
                StreamEvent clonedStreamEvent = streamEventCloner.copyStreamEvent(streamEvent);
                if (maxByMinByType.equals(MaxByMinByConstants.MIN_BY)) {
                    state.currentEvent = MaxByMinByExecutor
                            .getMinEventBatchProcessor(clonedStreamEvent, state.currentEvent, sortByAttribute);
                } else if (maxByMinByType.equals(MaxByMinByConstants.MAX_BY)) {
                    state.currentEvent = MaxByMinByExecutor
                            .getMaxEventBatchProcessor(clonedStreamEvent, state.currentEvent, sortByAttribute);
                }
            }
            streamEventChunk.clear();
            if (sendEvents) {
                if (state.expiredEventChunk.getFirst() != null) {
                    while (state.expiredEventChunk.hasNext()) {
                        StreamEvent expiredEvent = state.expiredEventChunk.next();
                        expiredEvent.setTimestamp(currentTime);
                    }
                    if (outputExpectsExpiredEvents) {
                        streamEventChunk.add(state.expiredEventChunk.getFirst());
                    }
                    state.resetEvent = streamEventCloner.copyStreamEvent(state.expiredEventChunk.getFirst());
                    if (state.resetEvent != null) {
                        state.resetEvent.setType(ComplexEvent.Type.RESET);
                        streamEventChunk.add(state.resetEvent);
                    }
                    state.resetEvent = null;
                }
                if (state.expiredEventChunk != null) {
                    state.expiredEventChunk.clear();
                }
                if (state.currentEvent != null) {
                    StreamEvent toExpireEvent = streamEventCloner.copyStreamEvent(state.currentEvent);
                    toExpireEvent.setType(StreamEvent.Type.EXPIRED);
                    state.expiredEventChunk.add(toExpireEvent);
                    streamEventChunk.add(state.currentEvent);
                }
                state.currentEvent = null;
            }
        }
        if (streamEventChunk.getFirst() != null) {
            streamEventChunk.setBatch(true);
            nextProcessor.process(streamEventChunk);
            streamEventChunk.setBatch(false);
        }
    }
@ Override protected void process ( ComplexEventChunk < StreamEvent > streamEventChunk , Processor nextProcessor , StreamEventCloner streamEventCloner , ExtensionState state ) { synchronized ( this ) { long currentTime = siddhiQueryContext . getSiddhiAppContext ( ) . getTimestampGenerator ( ) . currentTime ( ) ; if ( nextEmitTime == - 1 ) { if ( isStartTimeEnabled ) { nextEmitTime = getNextEmitTime ( currentTime ) ; } else { nextEmitTime = siddhiQueryContext . getSiddhiAppContext ( ) . getTimestampGenerator ( ) . currentTime ( ) + timeInMilliSeconds ; } scheduler . notifyAt ( nextEmitTime ) ; } boolean sendEvents ; if ( currentTime >= nextEmitTime ) { nextEmitTime += timeInMilliSeconds ; scheduler . notifyAt ( nextEmitTime ) ; sendEvents = true ; } else { sendEvents = false ; } while ( streamEventChunk . hasNext ( ) ) { StreamEvent streamEvent = streamEventChunk . next ( ) ; if ( streamEvent . getType ( ) != ComplexEvent . Type . CURRENT ) { continue ; } StreamEvent clonedStreamEvent = streamEventCloner . copyStreamEvent ( streamEvent ) ; if ( maxByMinByType . equals ( MaxByMinByConstants . MIN_BY ) ) { state . currentEvent = MaxByMinByExecutor . getMinEventBatchProcessor ( clonedStreamEvent , state . currentEvent , sortByAttribute ) ; } else if ( maxByMinByType . equals ( MaxByMinByConstants . MAX_BY ) ) { state . currentEvent = MaxByMinByExecutor . getMaxEventBatchProcessor ( clonedStreamEvent , state . currentEvent , sortByAttribute ) ; } } streamEventChunk . clear ( ) ; if ( sendEvents ) { if ( state . expiredEventChunk . getFirst ( ) != null ) { while ( state . expiredEventChunk . hasNext ( ) ) { StreamEvent expiredEvent = state . expiredEventChunk . next ( ) ; expiredEvent . setTimestamp ( currentTime ) ; } if ( outputExpectsExpiredEvents ) { streamEventChunk . add ( state . expiredEventChunk . getFirst ( ) ) ; } state . resetEvent = streamEventCloner . copyStreamEvent ( state . expiredEventChunk . getFirst ( ) ) ; if ( state . resetEvent != null ) { state . resetEvent . setType ( ComplexEvent . Type . RESET ) ; streamEventChunk . add ( state . resetEvent ) ; } state . resetEvent = null ; } if ( state . expiredEventChunk != null ) { state . expiredEventChunk . clear ( ) ; } if ( state . currentEvent != null ) { StreamEvent toExpireEvent = streamEventCloner . copyStreamEvent ( state . currentEvent ) ; toExpireEvent . setType ( StreamEvent . Type . EXPIRED ) ; state . expiredEventChunk . add ( toExpireEvent ) ; streamEventChunk . add ( state . currentEvent ) ; } state . currentEvent = null ; } } if ( streamEventChunk . getFirst ( ) != null ) { streamEventChunk . setBatch ( true ) ; nextProcessor . process ( streamEventChunk ) ; streamEventChunk . setBatch ( false ) ; } }
The main processing method that will be called upon event arrival
 @param streamEventChunk  the stream event chunk that need to be processed
 @param nextProcessor     the next processor to which the success events need to be passed
 @param streamEventCloner helps to clone the incoming event for local storage or modification
The main processing method that will be called upon event arrival @param streamEventChunk the stream event chunk that need to be processed @param nextProcessor the next processor to which the success events need to be passed @param streamEventCloner helps to clone the incoming event for local storage or modification
//for each event, set the min/max event as current event // add reset event before the current events