<commit_before>package io.sniffy;

import io.sniffy.configuration.SniffyConfiguration;
import io.sniffy.socket.*;
import io.sniffy.sql.SqlStats;
import io.sniffy.sql.StatementMetaData;
import io.sniffy.util.ExceptionUtil;
import io.sniffy.util.StringUtil;

import java.io.Closeable;
import java.lang.ref.WeakReference;
import java.util.*;
import java.util.concurrent.Callable;

import static io.sniffy.util.ExceptionUtil.throwException;

/**
 * Spy holds a number of queries which were executed at some point of time and uses it as a base for further assertions
 *
 * @see Sniffy#spy()
 * @see Sniffy#expect(Expectation)
 * @since 2.0
 */
public class Spy<C extends Spy<C>> extends LegacySpy<C> implements Closeable {

    private final WeakReference<Spy> selfReference;

    private boolean closed = false;
    private StackTraceElement[] closeStackTrace;

    private List<Expectation> expectations = new ArrayList<Expectation>();

    /**
     * @since 3.1
     */
    @Override
    public Map<StatementMetaData, SqlStats> getExecutedStatements(ThreadMatcher threadMatcher, boolean removeStackTraces) {

        Map<StatementMetaData, SqlStats> executedStatements = new LinkedHashMap<StatementMetaData, SqlStats>();
        for (Map.Entry<StatementMetaData, SqlStats> entry : this.executedStatements.ascendingMap().entrySet()) {

            StatementMetaData statementMetaData = entry.getKey();

            if (removeStackTraces) statementMetaData = new StatementMetaData(
                    statementMetaData.sql, statementMetaData.query, null, statementMetaData.getThreadMetaData()
            );

            if (threadMatcher.matches(statementMetaData.getThreadMetaData())) {
                SqlStats existingSocketStats = executedStatements.get(statementMetaData);
                if (null == existingSocketStats) {
                    executedStatements.put(statementMetaData, new SqlStats(entry.getValue()));
                } else {
                    existingSocketStats.accumulate(entry.getValue());
                }
            }
        }

        return Collections.unmodifiableMap(executedStatements);
    }

    Spy() {
        this(SpyConfiguration.builder().build());
    }

    Spy(SpyConfiguration spyConfiguration) {
        super(spyConfiguration);
        selfReference = Sniffy.registerSpy(this);
    }

    /**
     * Wrapper for {@link Sniffy#spy()} method; useful for chaining
     *
     * @return a new {@link Spy} instance
     * @since 2.0
     */
    public C reset() {
        checkOpened();
        super.reset();
        expectations.clear();
        return self();
    }

    /**
     * @since 3.1
     */
    public Map<SocketMetaData, SocketStats> getSocketOperations(ThreadMatcher threadMatcher, String address, boolean removeStackTraces) {
        return getSocketOperations(threadMatcher, AddressMatchers.exactAddressMatcher(address), removeStackTraces);
    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, SocketStats> getSocketOperations(ThreadMatcher threadMatcher, boolean removeStackTraces) {
        return getSocketOperations(threadMatcher, AddressMatchers.anyAddressMatcher(), removeStackTraces);
    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, SocketStats> getSocketOperations(ThreadMatcher threadMatcher, AddressMatcher addressMatcher, boolean removeStackTraces) {

        Map<SocketMetaData, SocketStats> socketOperations = new LinkedHashMap<SocketMetaData, SocketStats>();
        for (Map.Entry<SocketMetaData, SocketStats> entry : this.socketOperations.ascendingMap().entrySet()) {
            SocketMetaData socketMetaData = entry.getKey();
            if (threadMatcher.matches(socketMetaData.getThreadMetaData()) && (null == addressMatcher || addressMatcher.matches(socketMetaData.getAddress()))) {
                if (removeStackTraces) socketMetaData = new SocketMetaData(
                        socketMetaData.getProtocol(), socketMetaData.address, socketMetaData.connectionId, null, socketMetaData.getThreadMetaData()
                );
                SocketStats existingSocketStats = socketOperations.get(socketMetaData);
                if (null == existingSocketStats) {
                    socketOperations.put(socketMetaData, new SocketStats(entry.getValue()));
                } else {
                    existingSocketStats.accumulate(entry.getValue());
                }
            }
        }

        return Collections.unmodifiableMap(socketOperations);

    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, List<NetworkPacket>> getNetworkTraffic() {
        return getNetworkTraffic(Threads.ANY, AddressMatchers.anyAddressMatcher());
    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, List<NetworkPacket>> getNetworkTraffic(ThreadMatcher threadMatcher, String address) {
        return getNetworkTraffic(threadMatcher, AddressMatchers.exactAddressMatcher(address));
    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, List<NetworkPacket>> getNetworkTraffic(ThreadMatcher threadMatcher, String address, GroupingOptions groupingOptions) {
        return getNetworkTraffic(threadMatcher, AddressMatchers.exactAddressMatcher(address), groupingOptions);
    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, List<NetworkPacket>> getNetworkTraffic(ThreadMatcher threadMatcher, AddressMatcher addressMatcher) {
        return getNetworkTraffic(threadMatcher, addressMatcher, GroupingOptions.builder().build());
    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, List<NetworkPacket>> getNetworkTraffic(ThreadMatcher threadMatcher, AddressMatcher addressMatcher, GroupingOptions groupingOptions) {

        Map<SocketMetaData, List<NetworkPacket>> reducedTraffic = new LinkedHashMap<SocketMetaData, List<NetworkPacket>>();

        for (Map.Entry<SocketMetaData, Deque<NetworkPacket>> entry : this.networkTraffic.ascendingMap().entrySet()) {

            SocketMetaData socketMetaData = entry.getKey();
            Deque<NetworkPacket> networkPackets = entry.getValue();

            if (addressMatcher.matches(socketMetaData.getAddress()) &&
                    (null == socketMetaData.getThreadMetaData() || threadMatcher.matches(socketMetaData.getThreadMetaData()))) {

                for (NetworkPacket networkPacket : networkPackets) {

                    if (threadMatcher.matches(networkPacket.getThreadMetaData())) {

                        SocketMetaData reducedSocketMetaData = new SocketMetaData(
                                socketMetaData.getProtocol(),
                                socketMetaData.getAddress(),
                                groupingOptions.isGroupByConnection() ? socketMetaData.getConnectionId() : -1,
                                groupingOptions.isGroupByStackTrace() ? networkPacket.getStackTrace() : null,
                                groupingOptions.isGroupByThread() ? networkPacket.getThreadMetaData() : null
                        );

                        List<NetworkPacket> reducedNetworkPackets = reducedTraffic.get(reducedSocketMetaData);
                        //noinspection Java8MapApi
                        if (null == reducedNetworkPackets) {
                            reducedNetworkPackets = new ArrayList<NetworkPacket>();
                            reducedTraffic.put(reducedSocketMetaData, reducedNetworkPackets);
                        }

                        if (getSpyConfiguration().isCaptureStackTraces() && !groupingOptions.isGroupByStackTrace()
                                || groupingOptions.isGroupByThread()) {
                            byte[] bytes = networkPacket.getBytes();
                            networkPacket = new NetworkPacket(
                                    networkPacket.isSent(),
                                    networkPacket.getTimestamp(),
                                    groupingOptions.isGroupByStackTrace() ? networkPacket.getStackTrace() : null,
                                    groupingOptions.isGroupByThread() ? networkPacket.getThreadMetaData() : null,
                                    bytes, 0, bytes.length
                            );
                        }

                        if (reducedNetworkPackets.isEmpty()) {
                            reducedNetworkPackets.add(networkPacket);
                        } else {
                            NetworkPacket lastPacket = reducedNetworkPackets.get(reducedNetworkPackets.size() - 1);
                            if (!lastPacket.combine(networkPacket, SniffyConfiguration.INSTANCE.getPacketMergeThreshold())) {
                                reducedNetworkPackets.add(networkPacket);
                            }
                        }

                    }

                }

            }

        }<commit_msg>issues/424 Race condition in traffic capture - if packets are sent within same millisecond they are incorrectly merged<commit_after>package io.sniffy;

import io.sniffy.configuration.SniffyConfiguration;
import io.sniffy.socket.*;
import io.sniffy.sql.SqlStats;
import io.sniffy.sql.StatementMetaData;
import io.sniffy.util.ExceptionUtil;
import io.sniffy.util.StringUtil;

import java.io.Closeable;
import java.lang.ref.WeakReference;
import java.util.*;
import java.util.concurrent.Callable;

import static io.sniffy.util.ExceptionUtil.throwException;

/**
 * Spy holds a number of queries which were executed at some point of time and uses it as a base for further assertions
 *
 * @see Sniffy#spy()
 * @see Sniffy#expect(Expectation)
 * @since 2.0
 */
public class Spy<C extends Spy<C>> extends LegacySpy<C> implements Closeable {

    private final WeakReference<Spy> selfReference;

    private boolean closed = false;
    private StackTraceElement[] closeStackTrace;

    private List<Expectation> expectations = new ArrayList<Expectation>();

    /**
     * @since 3.1
     */
    @Override
    public Map<StatementMetaData, SqlStats> getExecutedStatements(ThreadMatcher threadMatcher, boolean removeStackTraces) {

        Map<StatementMetaData, SqlStats> executedStatements = new LinkedHashMap<StatementMetaData, SqlStats>();
        for (Map.Entry<StatementMetaData, SqlStats> entry : this.executedStatements.ascendingMap().entrySet()) {

            StatementMetaData statementMetaData = entry.getKey();

            if (removeStackTraces) statementMetaData = new StatementMetaData(
                    statementMetaData.sql, statementMetaData.query, null, statementMetaData.getThreadMetaData()
            );

            if (threadMatcher.matches(statementMetaData.getThreadMetaData())) {
                SqlStats existingSocketStats = executedStatements.get(statementMetaData);
                if (null == existingSocketStats) {
                    executedStatements.put(statementMetaData, new SqlStats(entry.getValue()));
                } else {
                    existingSocketStats.accumulate(entry.getValue());
                }
            }
        }

        return Collections.unmodifiableMap(executedStatements);
    }

    Spy() {
        this(SpyConfiguration.builder().build());
    }

    Spy(SpyConfiguration spyConfiguration) {
        super(spyConfiguration);
        selfReference = Sniffy.registerSpy(this);
    }

    /**
     * Wrapper for {@link Sniffy#spy()} method; useful for chaining
     *
     * @return a new {@link Spy} instance
     * @since 2.0
     */
    public C reset() {
        checkOpened();
        super.reset();
        expectations.clear();
        return self();
    }

    /**
     * @since 3.1
     */
    public Map<SocketMetaData, SocketStats> getSocketOperations(ThreadMatcher threadMatcher, String address, boolean removeStackTraces) {
        return getSocketOperations(threadMatcher, AddressMatchers.exactAddressMatcher(address), removeStackTraces);
    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, SocketStats> getSocketOperations(ThreadMatcher threadMatcher, boolean removeStackTraces) {
        return getSocketOperations(threadMatcher, AddressMatchers.anyAddressMatcher(), removeStackTraces);
    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, SocketStats> getSocketOperations(ThreadMatcher threadMatcher, AddressMatcher addressMatcher, boolean removeStackTraces) {

        Map<SocketMetaData, SocketStats> socketOperations = new LinkedHashMap<SocketMetaData, SocketStats>();
        for (Map.Entry<SocketMetaData, SocketStats> entry : this.socketOperations.ascendingMap().entrySet()) {
            SocketMetaData socketMetaData = entry.getKey();
            if (threadMatcher.matches(socketMetaData.getThreadMetaData()) && (null == addressMatcher || addressMatcher.matches(socketMetaData.getAddress()))) {
                if (removeStackTraces) socketMetaData = new SocketMetaData(
                        socketMetaData.getProtocol(), socketMetaData.address, socketMetaData.connectionId, null, socketMetaData.getThreadMetaData()
                );
                SocketStats existingSocketStats = socketOperations.get(socketMetaData);
                if (null == existingSocketStats) {
                    socketOperations.put(socketMetaData, new SocketStats(entry.getValue()));
                } else {
                    existingSocketStats.accumulate(entry.getValue());
                }
            }
        }

        return Collections.unmodifiableMap(socketOperations);

    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, List<NetworkPacket>> getNetworkTraffic() {
        return getNetworkTraffic(Threads.ANY, AddressMatchers.anyAddressMatcher());
    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, List<NetworkPacket>> getNetworkTraffic(ThreadMatcher threadMatcher, String address) {
        return getNetworkTraffic(threadMatcher, AddressMatchers.exactAddressMatcher(address));
    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, List<NetworkPacket>> getNetworkTraffic(ThreadMatcher threadMatcher, String address, GroupingOptions groupingOptions) {
        return getNetworkTraffic(threadMatcher, AddressMatchers.exactAddressMatcher(address), groupingOptions);
    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, List<NetworkPacket>> getNetworkTraffic(ThreadMatcher threadMatcher, AddressMatcher addressMatcher) {
        return getNetworkTraffic(threadMatcher, addressMatcher, GroupingOptions.builder().build());
    }

    /**
     * @since 3.1.10
     */
    public Map<SocketMetaData, List<NetworkPacket>> getNetworkTraffic(ThreadMatcher threadMatcher, AddressMatcher addressMatcher, GroupingOptions groupingOptions) {

        Map<SocketMetaData, List<NetworkPacket>> reducedTraffic = new LinkedHashMap<SocketMetaData, List<NetworkPacket>>();

        for (Map.Entry<SocketMetaData, Deque<NetworkPacket>> entry : this.networkTraffic.ascendingMap().entrySet()) {

            SocketMetaData socketMetaData = entry.getKey();
            Deque<NetworkPacket> networkPackets = entry.getValue();

            if (addressMatcher.matches(socketMetaData.getAddress()) &&
                    (null == socketMetaData.getThreadMetaData() || threadMatcher.matches(socketMetaData.getThreadMetaData()))) {

                for (NetworkPacket networkPacket : networkPackets) {

                    if (threadMatcher.matches(networkPacket.getThreadMetaData())) {

                        SocketMetaData reducedSocketMetaData = new SocketMetaData(
                                socketMetaData.getProtocol(),
                                socketMetaData.getAddress(),
                                groupingOptions.isGroupByConnection() ? socketMetaData.getConnectionId() : -1,
                                groupingOptions.isGroupByStackTrace() ? networkPacket.getStackTrace() : null,
                                groupingOptions.isGroupByThread() ? networkPacket.getThreadMetaData() : null
                        );

                        List<NetworkPacket> reducedNetworkPackets = reducedTraffic.get(reducedSocketMetaData);
                        //noinspection Java8MapApi
                        if (null == reducedNetworkPackets) {
                            reducedNetworkPackets = new ArrayList<NetworkPacket>();
                            reducedTraffic.put(reducedSocketMetaData, reducedNetworkPackets);
                        }

                        if ((getSpyConfiguration().isCaptureStackTraces() && !groupingOptions.isGroupByStackTrace())
                                || !groupingOptions.isGroupByThread()) {
                            byte[] bytes = networkPacket.getBytes();
                            networkPacket = new NetworkPacket(
                                    networkPacket.isSent(),
                                    networkPacket.getTimestamp(),
                                    groupingOptions.isGroupByStackTrace() ? networkPacket.getStackTrace() : null,
                                    groupingOptions.isGroupByThread() ? networkPacket.getThreadMetaData() : null,
                                    bytes, 0, bytes.length
                            );
                        }

                        if (reducedNetworkPackets.isEmpty()) {
                            reducedNetworkPackets.add(networkPacket);
                        } else {
                            NetworkPacket lastPacket = reducedNetworkPackets.get(reducedNetworkPackets.size() - 1);
                            if (!lastPacket.combine(networkPacket, SniffyConfiguration.INSTANCE.getPacketMergeThreshold())) {
                                reducedNetworkPackets.add(networkPacket);
                            }
                        }

                    }

                }

            }

        }
{'fraction_non_alphanumeric': 0.09195337890843887, 'fraction_numerical': 0.004650902162949681, 'mean_word_length': 2.3572234762979685, 'pattern_counts': {'":': 0, '<': 73, '<?xml version=': 0, '>': 73, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '664945', 'n_tokens_mistral': 4359, 'n_tokens_neox': 4265, 'n_words': 792}