@Override
    protected void processEventChunk(ComplexEventChunk<StreamEvent> streamEventChunk, Processor nextProcessor,
                                     StreamEventCloner streamEventCloner, GroupingKeyPopulator groupingKeyPopulater,
                                     WindowState state) {

        String key = DEFAULT_KEY;
        SessionComplexEventChunk<StreamEvent> currentSession;

        synchronized (state) {
            while (streamEventChunk.hasNext()) {
                StreamEvent streamEvent = streamEventChunk.next();
                long eventTimestamp = streamEvent.getTimestamp();
                long maxTimestamp = eventTimestamp + sessionGap;
                long aliveTimestamp = maxTimestamp + allowedLatency;

                if (streamEvent.getType() == StreamEvent.Type.CURRENT) {

                    if (sessionKeyExecutor != null) {
                        key = (String) sessionKeyExecutor.execute(streamEvent);
                    }

                    //Update event with the grouping key
                    groupingKeyPopulater.populateComplexEvent(streamEvent, key);


                    //get the session configuration based on session key
                    //if the map doesn't contain key, then a new sessionContainer
                    //object needs to be created.
                    if ((state.sessionContainer = state.sessionMap.get(key)) == null) {
                        state.sessionContainer = new SessionContainer(key);
                    }
                    state.sessionMap.put(key, state.sessionContainer);

                    StreamEvent clonedStreamEvent = streamEventCloner.copyStreamEvent(streamEvent);
                    clonedStreamEvent.setType(StreamEvent.Type.EXPIRED);

                    currentSession = state.sessionContainer.getCurrentSession();

                    //if current session is empty
                    if (state.sessionContainer.getCurrentSession().getFirst() == null) {
                        currentSession.add(clonedStreamEvent);
                        currentSession.setTimestamps(eventTimestamp, maxTimestamp, aliveTimestamp);
                        scheduler.notifyAt(maxTimestamp);
                    } else {
                        if (eventTimestamp >= currentSession.getStartTimestamp()) {
                            //check whether the event belongs to the same session
                            if (eventTimestamp <= currentSession.getEndTimestamp()) {
                                currentSession.setTimestamps(currentSession.getStartTimestamp(),
                                        maxTimestamp, aliveTimestamp);
                                currentSession.add(clonedStreamEvent);
                                scheduler.notifyAt(maxTimestamp);
                            } else {
                                //when a new session starts
                                if (allowedLatency > 0) {
                                    moveCurrentSessionToPreviousSession(state);
                                    currentSession.clear();
                                    currentSession.setTimestamps(eventTimestamp, maxTimestamp, aliveTimestamp);
                                    currentSession.add(clonedStreamEvent);
                                    scheduler.notifyAt(maxTimestamp);
                                }
                            }

                        } else {
                            //when a late event arrives
                            addLateEvent(streamEventChunk, eventTimestamp, clonedStreamEvent, state);
                        }
                    }
                } else {
                    currentSessionTimeout(eventTimestamp, state);
                    if (allowedLatency > 0) {
                        previousSessionTimeout(eventTimestamp, state);
                    }
                }
            }
        }

        if (state.expiredEventChunk != null && state.expiredEventChunk.getFirst() != null) {
            streamEventChunk.add((StreamEvent) state.expiredEventChunk.getFirst());
            state.expiredEventChunk.clear();
        }

        nextProcessor.process(streamEventChunk);

    }
Implementation of {@link WindowProcessor} which represent a Window operating based on a session.
modules/siddhi-core/src/main/java/io/siddhi/core/query/processor/stream/window/SessionWindowProcessor.java