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

        synchronized (state) {
            long currentTime = siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime();

            StreamEvent streamEvent = streamEventChunk.getFirst();
            streamEventChunk.clear();
            while (streamEvent != null) {
                StreamEvent next = streamEvent.getNext();
                streamEvent.setNext(null);

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

                state.totalCount++;
                if (state.totalCount != 1) {
                    state.currentBucketId = Math.ceil(state.totalCount / windowWidth);
                }
                String currentKey = generateKey(streamEvent);
                StreamEvent oldEvent = state.map.put(currentKey, clonedEvent);
                if (oldEvent != null) {    // this event is already in the store
                    state.countMap.put(currentKey, state.countMap.get(currentKey).incrementCount());
                } else {
                    //  This is a new event
                    LossyCount lCount;
                    lCount = new LossyCount(1, (int) state.currentBucketId - 1);
                    state.countMap.put(currentKey, lCount);
                }
                // calculating all the events in the system which match the
                // requirement provided by the user
                List<String> keys = new ArrayList<String>();
                keys.addAll(state.countMap.keySet());
                for (String key : keys) {
                    LossyCount lossyCount = state.countMap.get(key);
                    if (lossyCount.getCount() >= ((support - error) * state.totalCount)) {
                        // among the selected events, if the newly arrive event is there we mark it as an inEvent
                        if (key.equals(currentKey)) {
                            streamEventChunk.add(streamEvent);
                        }
                    }
                }
                if (state.totalCount % windowWidth == 0) {
                    // its time to run the data-structure prune code
                    keys = new ArrayList<String>();
                    keys.addAll(state.countMap.keySet());
                    for (String key : keys) {
                        LossyCount lossyCount = state.countMap.get(key);
                        if (lossyCount.getCount() + lossyCount.getBucketId() <= state.currentBucketId) {
                            log.info("Removing the Event: " + key + " from the window");
                            state.countMap.remove(key);
                            StreamEvent expirtedEvent = state.map.remove(key);
                            expirtedEvent.setTimestamp(currentTime);
                            streamEventChunk.add(expirtedEvent);
                        }
                    }
                }
                streamEvent = next;
            }
        }
        nextProcessor.process(streamEventChunk);
    }
Implementation of {@link WindowProcessor} which represent a Window operating based on event frequency.
modules/siddhi-core/src/main/java/io/siddhi/core/query/processor/stream/window/LossyFrequentWindowProcessor.java