component/src/main/java/org/wso2/extension/siddhi/execution/extrema/MaxByMinByLengthWindowProcessor.java
@Override
    protected void process(ComplexEventChunk<StreamEvent> streamEventChunk, Processor nextProcessor,
                           StreamEventCloner streamEventCloner, ExtensionState state) {
        List<ComplexEventChunk<StreamEvent>> streamEventChunks = new ArrayList<ComplexEventChunk<StreamEvent>>();
        synchronized (this) {
            long currentTime = siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime();
            while (streamEventChunk.hasNext()) {
                StreamEvent streamEvent = streamEventChunk.next();
                StreamEvent clonedStreamEvent = streamEventCloner
                        .copyStreamEvent(streamEvent); //this cloned event used to
                // insert the event into treemap

                if (state.count != 0) {
                    outputStreamEventChunk.clear();
                    internalWindowChunk.clear();
                }

                //get the parameter value for every events
                Object parameterValue = getParameterValue(minByMaxByExecutorAttribute, streamEvent);

                //insert the cloned event into treemap
                maxByMinByExecutor.insert(clonedStreamEvent, parameterValue);

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

                    //get the output event
                    setOutputStreamEvent(maxByMinByExecutor.getResult(maxByMinByExecutor.getMinByMaxByExecutorType()));

                    if (state.toBeExpiredEvent != null) {
                        if (outputStreamEvent != state.toBeExpiredEvent) {
                            state.toBeExpiredEvent.setTimestamp(currentTime);
                            state.toBeExpiredEvent.setType(StateEvent.Type.EXPIRED);
                            outputStreamEventChunk.add(state.toBeExpiredEvent);
                        }
                    }

                    outputStreamEventChunk.add(outputStreamEvent);
                    //add the event which is to be expired
                    internalWindowChunk.add(streamEventCloner.copyStreamEvent(outputStreamEvent));
                    state.toBeExpiredEvent = outputStreamEvent;
                    if (outputStreamEventChunk.getFirst() != null) {
                        streamEventChunks.add(outputStreamEventChunk);
                    }

                    events.add(clonedStreamEvent);
                } else {
                    StreamEvent firstEvent = events.get(0);
                    if (firstEvent != null) {
                        firstEvent.setTimestamp(currentTime);

                        //remove the expired event from treemap
                        Object expiredEventParameterValue = getParameterValue(minByMaxByExecutorAttribute, firstEvent);

                        maxByMinByExecutor.getSortedEventMap().remove(expiredEventParameterValue);
                        events.remove(0);

                        //get the output event
                        setOutputStreamEvent(
                                maxByMinByExecutor.getResult(maxByMinByExecutor.getMinByMaxByExecutorType()));
                        if (state.toBeExpiredEvent != null) {
                            if (outputStreamEvent != state.toBeExpiredEvent) {
                                state.toBeExpiredEvent.setTimestamp(currentTime);
                                state.toBeExpiredEvent.setType(StateEvent.Type.EXPIRED);
                                outputStreamEventChunk.add(state.toBeExpiredEvent);
                            }

                        }
                        outputStreamEventChunk.add(outputStreamEvent);
                        internalWindowChunk.add(streamEventCloner.copyStreamEvent(outputStreamEvent));
                        state.toBeExpiredEvent = outputStreamEvent;

                        if (outputStreamEventChunk.getFirst() != null) {
                            streamEventChunks.add(outputStreamEventChunk);
                        }
                        events.add(clonedStreamEvent);

                    }
                }
            }
        }
        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) {
            long currentTime = siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime();
            while (streamEventChunk.hasNext()) {
                StreamEvent streamEvent = streamEventChunk.next();
                StreamEvent clonedStreamEvent = streamEventCloner
                        .copyStreamEvent(streamEvent); 
                if (state.count != 0) {
                    outputStreamEventChunk.clear();
                    internalWindowChunk.clear();
                }
                Object parameterValue = getParameterValue(minByMaxByExecutorAttribute, streamEvent);
                maxByMinByExecutor.insert(clonedStreamEvent, parameterValue);
                if (state.count < length) {
                    state.count++;
                    setOutputStreamEvent(maxByMinByExecutor.getResult(maxByMinByExecutor.getMinByMaxByExecutorType()));
                    if (state.toBeExpiredEvent != null) {
                        if (outputStreamEvent != state.toBeExpiredEvent) {
                            state.toBeExpiredEvent.setTimestamp(currentTime);
                            state.toBeExpiredEvent.setType(StateEvent.Type.EXPIRED);
                            outputStreamEventChunk.add(state.toBeExpiredEvent);
                        }
                    }
                    outputStreamEventChunk.add(outputStreamEvent);
                    internalWindowChunk.add(streamEventCloner.copyStreamEvent(outputStreamEvent));
                    state.toBeExpiredEvent = outputStreamEvent;
                    if (outputStreamEventChunk.getFirst() != null) {
                        streamEventChunks.add(outputStreamEventChunk);
                    }
                    events.add(clonedStreamEvent);
                } else {
                    StreamEvent firstEvent = events.get(0);
                    if (firstEvent != null) {
                        firstEvent.setTimestamp(currentTime);
                        Object expiredEventParameterValue = getParameterValue(minByMaxByExecutorAttribute, firstEvent);
                        maxByMinByExecutor.getSortedEventMap().remove(expiredEventParameterValue);
                        events.remove(0);
                        setOutputStreamEvent(
                                maxByMinByExecutor.getResult(maxByMinByExecutor.getMinByMaxByExecutorType()));
                        if (state.toBeExpiredEvent != null) {
                            if (outputStreamEvent != state.toBeExpiredEvent) {
                                state.toBeExpiredEvent.setTimestamp(currentTime);
                                state.toBeExpiredEvent.setType(StateEvent.Type.EXPIRED);
                                outputStreamEventChunk.add(state.toBeExpiredEvent);
                            }
                        }
                        outputStreamEventChunk.add(outputStreamEvent);
                        internalWindowChunk.add(streamEventCloner.copyStreamEvent(outputStreamEvent));
                        state.toBeExpiredEvent = outputStreamEvent;
                        if (outputStreamEventChunk.getFirst() != null) {
                            streamEventChunks.add(outputStreamEventChunk);
                        }
                        events.add(clonedStreamEvent);
                    }
                }
            }
        }
        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 ) { long currentTime = siddhiQueryContext . getSiddhiAppContext ( ) . getTimestampGenerator ( ) . currentTime ( ) ; while ( streamEventChunk . hasNext ( ) ) { StreamEvent streamEvent = streamEventChunk . next ( ) ; StreamEvent clonedStreamEvent = streamEventCloner . copyStreamEvent ( streamEvent ) ; if ( state . count != 0 ) { outputStreamEventChunk . clear ( ) ; internalWindowChunk . clear ( ) ; } Object parameterValue = getParameterValue ( minByMaxByExecutorAttribute , streamEvent ) ; maxByMinByExecutor . insert ( clonedStreamEvent , parameterValue ) ; if ( state . count < length ) { state . count ++ ; setOutputStreamEvent ( maxByMinByExecutor . getResult ( maxByMinByExecutor . getMinByMaxByExecutorType ( ) ) ) ; if ( state . toBeExpiredEvent != null ) { if ( outputStreamEvent != state . toBeExpiredEvent ) { state . toBeExpiredEvent . setTimestamp ( currentTime ) ; state . toBeExpiredEvent . setType ( StateEvent . Type . EXPIRED ) ; outputStreamEventChunk . add ( state . toBeExpiredEvent ) ; } } outputStreamEventChunk . add ( outputStreamEvent ) ; internalWindowChunk . add ( streamEventCloner . copyStreamEvent ( outputStreamEvent ) ) ; state . toBeExpiredEvent = outputStreamEvent ; if ( outputStreamEventChunk . getFirst ( ) != null ) { streamEventChunks . add ( outputStreamEventChunk ) ; } events . add ( clonedStreamEvent ) ; } else { StreamEvent firstEvent = events . get ( 0 ) ; if ( firstEvent != null ) { firstEvent . setTimestamp ( currentTime ) ; Object expiredEventParameterValue = getParameterValue ( minByMaxByExecutorAttribute , firstEvent ) ; maxByMinByExecutor . getSortedEventMap ( ) . remove ( expiredEventParameterValue ) ; events . remove ( 0 ) ; setOutputStreamEvent ( maxByMinByExecutor . getResult ( maxByMinByExecutor . getMinByMaxByExecutorType ( ) ) ) ; if ( state . toBeExpiredEvent != null ) { if ( outputStreamEvent != state . toBeExpiredEvent ) { state . toBeExpiredEvent . setTimestamp ( currentTime ) ; state . toBeExpiredEvent . setType ( StateEvent . Type . EXPIRED ) ; outputStreamEventChunk . add ( state . toBeExpiredEvent ) ; } } outputStreamEventChunk . add ( outputStreamEvent ) ; internalWindowChunk . add ( streamEventCloner . copyStreamEvent ( outputStreamEvent ) ) ; state . toBeExpiredEvent = outputStreamEvent ; if ( outputStreamEventChunk . getFirst ( ) != null ) { streamEventChunks . add ( outputStreamEventChunk ) ; } events . add ( clonedStreamEvent ) ; } } } } 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
//this cloned event used to // insert the event into treemap //get the parameter value for every events //insert the cloned event into treemap //get the output event //add the event which is to be expired //remove the expired event from treemap //get the output event