@Override
    protected void process(ComplexEventChunk<StreamEvent> streamEventChunk, Processor nextProcessor,
                           StreamEventCloner streamEventCloner, WindowState state) {
        synchronized (state) {
            if (nextEmitTime == -1) {
                long currentTime = siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime();
                if (isStartTimeEnabled) {
                    nextEmitTime = getNextEmitTime(currentTime);
                } else {
                    nextEmitTime = siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime()
                            + timeInMilliSeconds;
                }
                scheduler.notifyAt(nextEmitTime);
            }
            long currentTime = siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime();
            boolean sendEvents;

            if (currentTime >= nextEmitTime) {
                nextEmitTime += timeInMilliSeconds;
                scheduler.notifyAt(nextEmitTime);
                sendEvents = true;
            } else {
                sendEvents = false;
            }

            while (streamEventChunk.hasNext()) {
                StreamEvent streamEvent = streamEventChunk.next();
                if (streamEvent.getType() != ComplexEvent.Type.CURRENT) {
                    continue;
                }
                StreamEvent clonedStreamEvent = streamEventCloner.copyStreamEvent(streamEvent);
                if (state.resetEvent == null) {
                    state.resetEvent = streamEventCloner.copyStreamEvent(streamEvent);
                    state.resetEvent.setType(ComplexEvent.Type.RESET);
                }
                if (!isStreamCurrentEvents) {
                    state.currentEventQueue.add(clonedStreamEvent);
                } else if (state.expiredEventQueue != null) {
                    clonedStreamEvent.setType(StreamEvent.Type.EXPIRED);
                    state.expiredEventQueue.add(clonedStreamEvent);
                }
            }
            if (!isStreamCurrentEvents) {
                streamEventChunk.clear();
            }
            if (sendEvents) {
                if (outputExpectsExpiredEvents && state.expiredEventQueue.getFirst() != null) {
                    while (state.expiredEventQueue.hasNext()) {
                        StreamEvent expiredEvent = state.expiredEventQueue.next();
                        expiredEvent.setTimestamp(currentTime);
                    }
                    streamEventChunk.add(state.expiredEventQueue.getFirst());
                    state.expiredEventQueue.clear();
                }

                if (state.resetEvent != null) {
                    streamEventChunk.add(state.resetEvent);
                    state.resetEvent = null;
                }

                if (state.currentEventQueue != null && state.currentEventQueue.getFirst() != null) {
                    if (state.expiredEventQueue != null) {
                        state.currentEventQueue.reset();
                        while (state.currentEventQueue.hasNext()) {
                            StreamEvent currentEvent = state.currentEventQueue.next();
                            StreamEvent toExpireEvent = streamEventCloner.copyStreamEvent(currentEvent);
                            toExpireEvent.setType(StreamEvent.Type.EXPIRED);
                            state.expiredEventQueue.add(toExpireEvent);
                        }
                    }
                    streamEventChunk.add(state.currentEventQueue.getFirst());
                    state.currentEventQueue.clear();
                }
            }
        }
        if (streamEventChunk.getFirst() != null) {
            nextProcessor.process(streamEventChunk);
        }
    }
Implementation of {@link WindowProcessor} which represent a Batch Window operating based on time.
modules/siddhi-core/src/main/java/io/siddhi/core/query/processor/stream/window/TimeBatchWindowProcessor.java