@Override
    protected void process(ComplexEventChunk<StreamEvent> streamEventChunk, Processor nextProcessor,
                           StreamEventCloner streamEventCloner, WindowState state) {
        synchronized (state) {
            long currentTime = siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime();
            while (streamEventChunk.hasNext()) {
                StreamEvent streamEvent = streamEventChunk.next();
                StreamEvent clonedEvent = streamEventCloner.copyStreamEvent(streamEvent);
                clonedEvent.setType(StreamEvent.Type.EXPIRED);
                if (state.count < length) {
                    state.count++;
                    state.expiredEventQueue.add(clonedEvent);
                } else {
                    StreamEvent firstEvent = state.expiredEventQueue.poll();
                    if (firstEvent != null) {
                        firstEvent.setTimestamp(currentTime);
                        streamEventChunk.insertBeforeCurrent(firstEvent);
                        state.expiredEventQueue.add(clonedEvent);
                    } else {
                        StreamEvent resetEvent = streamEventCloner.copyStreamEvent(streamEvent);
                        resetEvent.setType(ComplexEvent.Type.RESET);
                        // adding resetEvent and clonedEvent event to the streamEventChunk
                        // since we are using insertAfterCurrent(), the final order will be
                        // currentEvent > clonedEvent (or expiredEvent) > resetEvent
                        streamEventChunk.insertAfterCurrent(resetEvent);
                        streamEventChunk.insertAfterCurrent(clonedEvent);

                        // since we manually added resetEvent and clonedEvent in earlier step
                        // we have to skip those two events from getting processed in the next
                        // iteration. Hence, calling next() twice.
                        streamEventChunk.next();
                        streamEventChunk.next();
                    }
                }
            }
        }
        nextProcessor.process(streamEventChunk);
    }
Implementation of {@link WindowProcessor} which represent a Window operating based on a pre-defined length.
modules/siddhi-core/src/main/java/io/siddhi/core/query/processor/stream/window/LengthWindowProcessor.java