@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();

                state.expiredEventQueue.reset();
                while (state.expiredEventQueue.hasNext()) {
                    StreamEvent expiredEvent = state.expiredEventQueue.next();
                    long timeDiff = expiredEvent.getTimestamp() - currentTime + timeInMilliSeconds;
                    if (timeDiff <= 0) {
                        state.expiredEventQueue.remove();
                        state.count--;
                        expiredEvent.setTimestamp(currentTime);
                        streamEventChunk.insertBeforeCurrent(expiredEvent);
                    } else {
                        break;
                    }
                }

                state.expiredEventQueue.reset();

                if (streamEvent.getType() == StreamEvent.Type.CURRENT) {
                    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);
                        }
                    }
                    scheduler.notifyAt(clonedEvent.getTimestamp() + timeInMilliSeconds);
                } else {
                    streamEventChunk.remove();
                }
            }
        }
        streamEventChunk.reset();
        if (streamEventChunk.hasNext()) {
            nextProcessor.process(streamEventChunk);
        }
    }
Implementation of {@link WindowProcessor} which represent a Window operating based on pre-defined length.
modules/siddhi-core/src/main/java/io/siddhi/core/query/processor/stream/window/TimeLengthWindowProcessor.java