component/src/main/java/org/wso2/extension/siddhi/execution/extrema/MaxByMinByLengthBatchWindowProcessor.java
@Override
    protected void process(ComplexEventChunk<StreamEvent> streamEventChunk, Processor nextProcessor,
                           StreamEventCloner streamEventCloner, ExtensionState state) {
        List<ComplexEventChunk<StreamEvent>> streamEventChunks = new ArrayList<ComplexEventChunk<StreamEvent>>();
        synchronized (this) {
            ComplexEventChunk<StreamEvent> outputStreamEventChunk = new ComplexEventChunk<StreamEvent>(true);
            long currentTime = this.siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime();
            while (streamEventChunk.hasNext()) {
                StreamEvent streamEvent = streamEventChunk.next();
                StreamEvent currentEvent = streamEventCloner.copyStreamEvent(streamEvent);

                if (state.count == 0) {
                    outputStreamEventChunk.clear();
                    oldEvent = null;
                    //clonedResultEvent=resultEvent;
                }

                //Get the event which hold the minimum or maximum event

                if (minByMaxByExecutorType.equals(MaxByMinByConstants.MAX_BY)) {
                    state.resultEvent = MaxByMinByExecutor
                            .getMaxEventBatchProcessor(currentEvent, oldEvent, minByMaxByExecutorAttribute);
                    oldEvent = state.resultEvent;
                } else if (minByMaxByExecutorType.equals(MaxByMinByConstants.MIN_BY)) {
                    state.resultEvent = MaxByMinByExecutor
                            .getMinEventBatchProcessor(currentEvent, oldEvent, minByMaxByExecutorAttribute);
                    oldEvent = state.resultEvent;
                }

                state.count++;
                if (state.count == length) {

                    if (state.resultEvent != null) {
                        if (expiredResultEvent != null) {
                            state.expiredEventChunk.clear();
                            outputStreamEventChunk.add(expiredResultEvent);
                            outputStreamEventChunk.add(state.resetEvent);

                        }
                        outputStreamEventChunk.add(state.resultEvent);
                        expiredResultEvent = streamEventCloner.copyStreamEvent(state.resultEvent);
                        expiredResultEvent.setTimestamp(currentTime);
                        expiredResultEvent.setType(StreamEvent.Type.EXPIRED);
                        state.expiredEventChunk.add(expiredResultEvent);
                        state.resetEvent = streamEventCloner.copyStreamEvent(state.resultEvent);
                        state.resetEvent.setType(StateEvent.Type.RESET);
                    }
                    state.count = 0;
                    if (outputStreamEventChunk.getFirst() != null) {
                        streamEventChunks.add(outputStreamEventChunk);
                    }
                }

            }
        }
        for (ComplexEventChunk<StreamEvent> outputStreamEventChunk : streamEventChunks) {
            nextProcessor.process(outputStreamEventChunk);
        }
    }
/**
     * 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) {
        List<ComplexEventChunk<StreamEvent>> streamEventChunks = new ArrayList<ComplexEventChunk<StreamEvent>>();
        synchronized (this) {
            ComplexEventChunk<StreamEvent> outputStreamEventChunk = new ComplexEventChunk<StreamEvent>(true);
            long currentTime = this.siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime();
            while (streamEventChunk.hasNext()) {
                StreamEvent streamEvent = streamEventChunk.next();
                StreamEvent currentEvent = streamEventCloner.copyStreamEvent(streamEvent);
                if (state.count == 0) {
                    outputStreamEventChunk.clear();
                    oldEvent = null;
                }
                if (minByMaxByExecutorType.equals(MaxByMinByConstants.MAX_BY)) {
                    state.resultEvent = MaxByMinByExecutor
                            .getMaxEventBatchProcessor(currentEvent, oldEvent, minByMaxByExecutorAttribute);
                    oldEvent = state.resultEvent;
                } else if (minByMaxByExecutorType.equals(MaxByMinByConstants.MIN_BY)) {
                    state.resultEvent = MaxByMinByExecutor
                            .getMinEventBatchProcessor(currentEvent, oldEvent, minByMaxByExecutorAttribute);
                    oldEvent = state.resultEvent;
                }
                state.count++;
                if (state.count == length) {
                    if (state.resultEvent != null) {
                        if (expiredResultEvent != null) {
                            state.expiredEventChunk.clear();
                            outputStreamEventChunk.add(expiredResultEvent);
                            outputStreamEventChunk.add(state.resetEvent);
                        }
                        outputStreamEventChunk.add(state.resultEvent);
                        expiredResultEvent = streamEventCloner.copyStreamEvent(state.resultEvent);
                        expiredResultEvent.setTimestamp(currentTime);
                        expiredResultEvent.setType(StreamEvent.Type.EXPIRED);
                        state.expiredEventChunk.add(expiredResultEvent);
                        state.resetEvent = streamEventCloner.copyStreamEvent(state.resultEvent);
                        state.resetEvent.setType(StateEvent.Type.RESET);
                    }
                    state.count = 0;
                    if (outputStreamEventChunk.getFirst() != null) {
                        streamEventChunks.add(outputStreamEventChunk);
                    }
                }
            }
        }
        for (ComplexEventChunk<StreamEvent> outputStreamEventChunk : streamEventChunks) {
            nextProcessor.process(outputStreamEventChunk);
        }
    }
@ Override protected void process ( ComplexEventChunk < StreamEvent > streamEventChunk , Processor nextProcessor , StreamEventCloner streamEventCloner , ExtensionState state ) { List < ComplexEventChunk < StreamEvent > > streamEventChunks = new ArrayList < ComplexEventChunk < StreamEvent > > ( ) ; synchronized ( this ) { ComplexEventChunk < StreamEvent > outputStreamEventChunk = new ComplexEventChunk < StreamEvent > ( true ) ; long currentTime = this . siddhiQueryContext . getSiddhiAppContext ( ) . getTimestampGenerator ( ) . currentTime ( ) ; while ( streamEventChunk . hasNext ( ) ) { StreamEvent streamEvent = streamEventChunk . next ( ) ; StreamEvent currentEvent = streamEventCloner . copyStreamEvent ( streamEvent ) ; if ( state . count == 0 ) { outputStreamEventChunk . clear ( ) ; oldEvent = null ; } if ( minByMaxByExecutorType . equals ( MaxByMinByConstants . MAX_BY ) ) { state . resultEvent = MaxByMinByExecutor . getMaxEventBatchProcessor ( currentEvent , oldEvent , minByMaxByExecutorAttribute ) ; oldEvent = state . resultEvent ; } else if ( minByMaxByExecutorType . equals ( MaxByMinByConstants . MIN_BY ) ) { state . resultEvent = MaxByMinByExecutor . getMinEventBatchProcessor ( currentEvent , oldEvent , minByMaxByExecutorAttribute ) ; oldEvent = state . resultEvent ; } state . count ++ ; if ( state . count == length ) { if ( state . resultEvent != null ) { if ( expiredResultEvent != null ) { state . expiredEventChunk . clear ( ) ; outputStreamEventChunk . add ( expiredResultEvent ) ; outputStreamEventChunk . add ( state . resetEvent ) ; } outputStreamEventChunk . add ( state . resultEvent ) ; expiredResultEvent = streamEventCloner . copyStreamEvent ( state . resultEvent ) ; expiredResultEvent . setTimestamp ( currentTime ) ; expiredResultEvent . setType ( StreamEvent . Type . EXPIRED ) ; state . expiredEventChunk . add ( expiredResultEvent ) ; state . resetEvent = streamEventCloner . copyStreamEvent ( state . resultEvent ) ; state . resetEvent . setType ( StateEvent . Type . RESET ) ; } state . count = 0 ; if ( outputStreamEventChunk . getFirst ( ) != null ) { streamEventChunks . add ( outputStreamEventChunk ) ; } } } } for ( ComplexEventChunk < StreamEvent > outputStreamEventChunk : streamEventChunks ) { nextProcessor . process ( outputStreamEventChunk ) ; } }
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
//clonedResultEvent=resultEvent; //Get the event which hold the minimum or maximum event