<commit_before>/*
 * Copyright 2014 CyberVision, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.kaaproject.kaa.client.channel.impl.transports;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.kaaproject.kaa.client.channel.EventTransport;
import org.kaaproject.kaa.client.event.EventManager;
import org.kaaproject.kaa.client.persistence.KaaClientState;
import org.kaaproject.kaa.common.TransportType;
import org.kaaproject.kaa.common.endpoint.gen.Event;
import org.kaaproject.kaa.common.endpoint.gen.EventSequenceNumberRequest;
import org.kaaproject.kaa.common.endpoint.gen.EventSyncRequest;
import org.kaaproject.kaa.common.endpoint.gen.EventSyncResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DefaultEventTransport extends AbstractKaaTransport implements EventTransport
{
    private static final Logger LOG = LoggerFactory.getLogger(DefaultEventTransport.class);

    private final KaaClientState clientState;

    private final Map<Integer, Set<Event>> pendingEvents = new HashMap<Integer, Set<Event>>();
    private EventManager manager;
    private final EventComparator eventSeqNumberComparator = new EventComparator();

    private boolean isEventSNSynchronized = false;
    private int startEventSN;

    public DefaultEventTransport(KaaClientState state) {
        clientState = state;
        startEventSN = clientState.getEventSeqNum();
    }

    @Override
    public EventSyncRequest createEventRequest(Integer requestId) {
        if (manager != null) {
            EventSyncRequest request = new EventSyncRequest();

            manager.fillEventListenersSyncRequest(request);

            if (isEventSNSynchronized) {
                Set<Event> eventsSet = new HashSet<Event>();

                if (!pendingEvents.isEmpty()) {
                    for (Map.Entry<Integer, Set<Event>> pendingEvents : pendingEvents.entrySet()) {
                        LOG.debug("Have not received response for {} events sent with request id {}",
                                pendingEvents.getValue().size(), pendingEvents.getKey());
                        eventsSet.addAll(pendingEvents.getValue());
                    }
                }

                eventsSet.addAll(manager.getPendingEvents());

                List<Event> events = new ArrayList<Event>(eventsSet);
                if (!events.isEmpty()) {
                    Collections.sort(events, eventSeqNumberComparator);

                    if (events.get(0).getSeqNum() != startEventSN) {
                        clientState.setEventSeqNum(startEventSN + events.size());

                        LOG.info("Put in order event sequence numbers (expected: {}, actual: {})"
                                                        , startEventSN , events.get(0).getSeqNum());

                        for (Event e : events) {
                            e.setSeqNum(startEventSN++);
                        }
                    } else {
                        startEventSN += events.size();
                    }

                    LOG.debug("Going to send {} event{}", events.size()
                            , (events.size() == 1 ? "" : "s")); //NOSONAR
                    request.setEvents(events);
                    pendingEvents.put(requestId, eventsSet);
                }

                request.setEventSequenceNumberRequest(null);
            } else {
                request.setEventSequenceNumberRequest(new EventSequenceNumberRequest());
                LOG.trace("Sending event sequence number request: "
                                    + "restored_sn = {}", startEventSN);
            }

            return request;
        }
        return null;
    }

    @Override
    public void onEventResponse(EventSyncResponse response) {
        if (manager != null) {
            if (!isEventSNSynchronized && response.getEventSequenceNumberResponse() != null)
            {
                int lastSN = response.getEventSequenceNumberResponse().getSeqNum();
                int expectedSN = (lastSN > 0 ? lastSN + 1 : lastSN);

                if (startEventSN != expectedSN) {
                    startEventSN = expectedSN;
                    clientState.setEventSeqNum(startEventSN);
                    LOG.info("Event sequence number is unsynchronized. Set to {}", startEventSN);
                } else {
                    LOG.info("Event sequence number is up to date: {}", startEventSN);
                }

                isEventSNSynchronized = true;
            }

            if (response.getEvents() != null && !response.getEvents().isEmpty()) {
                List<Event> events = new ArrayList<>(response.getEvents());
                Collections.sort(events, eventSeqNumberComparator);
                for (Event event : events) {
                    manager.onGenericEvent(event.getEventClassFQN(), event.getEventData().array(), event.getSource());
                }
            }
            if (response.getEventListenersResponses() != null && !response.getEventListenersResponses().isEmpty()) {
                manager.eventListenersResponseReceived(response.getEventListenersResponses());
            }
        }
        LOG.trace("Processed event response");
    }

    @Override
    public void setEventManager(EventManager manager) {
        this.manager = manager;
    }

    class EventComparator implements Comparator<Event>{

        @Override
        public int compare(Event e1, Event e2) {
            return e1.getSeqNum() - e2.getSeqNum();
        }

    }

    @Override
    public void onSyncResposeIdReceived(Integer requestId) {
        LOG.debug("Events sent with request id {} were accepted.", requestId);
        Set<Event> acceptedEvents = pendingEvents.remove(requestId);
        Iterator<Entry<Integer, Set<Event>>> entrySetIterator = pendingEvents.entrySet().iterator();
        while(entrySetIterator.hasNext()){
            Entry<Integer, Set<Event>> entry = entrySetIterator.next();
            entry.getValue().removeAll(acceptedEvents);
            if(entry.getValue().size() == 0){
                LOG.debug("Remove entry for request {}.", requestId);
                entrySetIterator.remove();
            }<commit_msg>KAA-244: Improvement to pending events accounting<commit_after>/*
 * Copyright 2014 CyberVision, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.kaaproject.kaa.client.channel.impl.transports;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.kaaproject.kaa.client.channel.EventTransport;
import org.kaaproject.kaa.client.event.EventManager;
import org.kaaproject.kaa.client.persistence.KaaClientState;
import org.kaaproject.kaa.common.TransportType;
import org.kaaproject.kaa.common.endpoint.gen.Event;
import org.kaaproject.kaa.common.endpoint.gen.EventSequenceNumberRequest;
import org.kaaproject.kaa.common.endpoint.gen.EventSyncRequest;
import org.kaaproject.kaa.common.endpoint.gen.EventSyncResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DefaultEventTransport extends AbstractKaaTransport implements EventTransport {
    private static final Logger LOG = LoggerFactory.getLogger(DefaultEventTransport.class);

    private final KaaClientState clientState;

    private final Map<Integer, Set<Event>> pendingEvents = new HashMap<Integer, Set<Event>>();
    private EventManager manager;
    private final EventComparator eventSeqNumberComparator = new EventComparator();

    private boolean isEventSNSynchronized = false;
    private int startEventSN;

    public DefaultEventTransport(KaaClientState state) {
        clientState = state;
        startEventSN = clientState.getEventSeqNum();
    }

    @Override
    public EventSyncRequest createEventRequest(Integer requestId) {
        if (manager != null) {
            EventSyncRequest request = new EventSyncRequest();

            manager.fillEventListenersSyncRequest(request);

            if (isEventSNSynchronized) {
                Set<Event> eventsSet = new HashSet<Event>();

                if (!pendingEvents.isEmpty()) {
                    for (Map.Entry<Integer, Set<Event>> pendingEvents : pendingEvents.entrySet()) {
                        LOG.debug("Have not received response for {} events sent with request id {}", pendingEvents.getValue().size(),
                                pendingEvents.getKey());
                        eventsSet.addAll(pendingEvents.getValue());
                    }
                }

                eventsSet.addAll(manager.getPendingEvents());

                List<Event> events = new ArrayList<Event>(eventsSet);
                if (!events.isEmpty()) {
                    Collections.sort(events, eventSeqNumberComparator);
                    LOG.debug("Going to send {} event{}", events.size(), (events.size() == 1 ? "" : "s")); // NOSONAR
                    request.setEvents(events);
                    pendingEvents.put(requestId, eventsSet);
                }

                request.setEventSequenceNumberRequest(null);
            } else {
                request.setEventSequenceNumberRequest(new EventSequenceNumberRequest());
                LOG.trace("Sending event sequence number request: " + "restored_sn = {}", startEventSN);
            }

            return request;
        }
        return null;
    }

    @Override
    public void onEventResponse(EventSyncResponse response) {
        if (manager != null) {
            if (!isEventSNSynchronized && response.getEventSequenceNumberResponse() != null) {
                int lastSN = response.getEventSequenceNumberResponse().getSeqNum();
                int expectedSN = (lastSN > 0 ? lastSN + 1 : lastSN);

                if (startEventSN != expectedSN) {
                    startEventSN = expectedSN;
                    clientState.setEventSeqNum(startEventSN);

                    Set<Event> eventsSet = new HashSet<Event>();
                    for (Set<Event> events : pendingEvents.values()) {
                        eventsSet.addAll(events);
                    }

                    List<Event> events = new ArrayList<Event>(eventsSet);
                    Collections.sort(events, eventSeqNumberComparator);

                    clientState.setEventSeqNum(startEventSN + events.size());
                    if (events.get(0).getSeqNum() != startEventSN) {
                        LOG.info("Put in order event sequence numbers (expected: {}, actual: {})", startEventSN, events.get(0).getSeqNum());

                        for (Event e : events) {
                            e.setSeqNum(startEventSN++);
                        }
                    } else {
                        startEventSN += events.size();
                    }

                    LOG.info("Event sequence number is unsynchronized. Set to {}", startEventSN);
                } else {
                    LOG.info("Event sequence number is up to date: {}", startEventSN);
                }

                isEventSNSynchronized = true;
            }

            if (response.getEvents() != null && !response.getEvents().isEmpty()) {
                List<Event> events = new ArrayList<>(response.getEvents());
                Collections.sort(events, eventSeqNumberComparator);
                for (Event event : events) {
                    manager.onGenericEvent(event.getEventClassFQN(), event.getEventData().array(), event.getSource());
                }
            }
            if (response.getEventListenersResponses() != null && !response.getEventListenersResponses().isEmpty()) {
                manager.eventListenersResponseReceived(response.getEventListenersResponses());
            }
        }
        LOG.trace("Processed event response");
    }

    @Override
    public void setEventManager(EventManager manager) {
        this.manager = manager;
    }

    class EventComparator implements Comparator<Event> {

        @Override
        public int compare(Event e1, Event e2) {
            return e1.getSeqNum() - e2.getSeqNum();
        }

    }

    @Override
    public void onSyncResposeIdReceived(Integer requestId) {
        LOG.debug("Events sent with request id {} were accepted.", requestId);
        Set<Event> acceptedEvents = pendingEvents.remove(requestId);
        Iterator<Entry<Integer, Set<Event>>> entrySetIterator = pendingEvents.entrySet().iterator();
        while (entrySetIterator.hasNext()) {
            Entry<Integer, Set<Event>> entry = entrySetIterator.next();
            entry.getValue().removeAll(acceptedEvents);
            if (entry.getValue().size() == 0) {
                LOG.debug("Remove entry for request {}.", requestId);
                entrySetIterator.remove();
            }
{'fraction_non_alphanumeric': 0.10060717311493929, 'fraction_numerical': 0.0030358655746964133, 'mean_word_length': 2.2653296449976947, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 48, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2504187', 'n_tokens_mistral': 3596, 'n_tokens_neox': 3461, 'n_words': 818}