@Override
    protected void process(ComplexEventChunk<StreamEvent> streamEventChunk, Processor nextProcessor,
                           StreamEventCloner streamEventCloner, WindowState windowState) {
        DelayedWindowState state = ((DelayedWindowState) windowState);
        synchronized (state) {
            while (streamEventChunk.hasNext()) {
                StreamEvent streamEvent = streamEventChunk.next();
                long currentTime = siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime();

                state.delayedEventQueue.reset();
                while (state.delayedEventQueue.hasNext()) {
                    StreamEvent delayedEvent = state.delayedEventQueue.next();
                    long timeDiff = delayedEvent.getTimestamp() - currentTime + delayInMilliSeconds;
                    if (timeDiff <= 0) {
                        state.delayedEventQueue.remove();
                        //insert delayed event before the current event to stream chunk
                        delayedEvent.setTimestamp(currentTime);
                        streamEventChunk.insertBeforeCurrent(delayedEvent);
                    } else {
                        break;
                    }
                }

                if (streamEvent.getType() == StreamEvent.Type.CURRENT) {
                    state.delayedEventQueue.add(streamEvent);
                    if (state.lastTimestamp < streamEvent.getTimestamp()) {
                        getScheduler().notifyAt(streamEvent.getTimestamp() + delayInMilliSeconds);
                        state.lastTimestamp = streamEvent.getTimestamp();
                    }
                }
                //current events are not processed, so remove the current event from the stream chunk
                streamEventChunk.remove();
            }
            state.delayedEventQueue.reset();
        }
        //only pass to next processor if there are any events in the stream chunk
        if (streamEventChunk.getFirst() != null) {
            nextProcessor.process(streamEventChunk);
        }
    }
Implementation of {@link WindowProcessor} which represent a Window operating based on delay time.
modules/siddhi-core/src/main/java/io/siddhi/core/query/processor/stream/window/DelayWindowProcessor.java