<commit_before>package net.kencochrane.raven.connection;

import net.kencochrane.raven.Dsn;
import net.kencochrane.raven.event.Event;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Asynchronous usage of a connection.
 * <p>
 * Instead of synchronously sending each event to a connection, use a ThreadPool to establish the connection
 * and submit the event.
 * </p>
 */
public class AsyncConnection implements Connection {
    /**
     * DSN option for the number of threads assigned for the connection.
     */
    public static final String MAX_THREADS_OPTION = "raven.async.threads";
    /**
     * DSN option for the priority of threads assigned for the connection.
     */
    public static final String PRIORITY_OPTION = "raven.async.priority";
    private static final Logger logger = Logger.getLogger(AsyncConnection.class.getCanonicalName());
    /**
     * Timeout of the {@link #executorService}.
     */
    private static final int TIMEOUT = 1000;
    /**
     * Number of threads dedicated to the connection usage by default (Number of processors available).
     */
    private static final int DEFAULT_MAX_THREADS = Runtime.getRuntime().availableProcessors();
    /**
     * Default threads priority.
     */
    private static final int DEFAULT_PRIORITY = Thread.MIN_PRIORITY;
    /**
     * Connection used to actually send the events.
     */
    private final Connection actualConnection;
    /**
     * Executor service in charge of running the connection in separate threads.
     */
    private final ExecutorService executorService;
    /**
     * Option to disable the propagation of the {@link #close()} operation to the actual connection.
     */
    private final boolean propagateClose;

    /**
     * Creates a connection which will rely on an executor to send events.
     * <p>
     * Will propagate the {@link #close()} operation and use {@link #DEFAULT_MAX_THREADS} and {@link #DEFAULT_PRIORITY}.
     * </p>
     *
     * @param actualConnection connection used to send the events.
     */
    public AsyncConnection(Connection actualConnection) {
        this(actualConnection, true, DEFAULT_MAX_THREADS, DEFAULT_PRIORITY);
    }

    /**
     * Creates a connection which will rely on an executor to send events.
     * <p>
     * Will propagate the {@link #close()} operation and
     * </p>
     *
     * @param actualConnection
     * @param dsn
     */
    public AsyncConnection(Connection actualConnection, Dsn dsn) {
        this(actualConnection, true, getMaxThreads(dsn), getPriority(dsn));
    }

    public AsyncConnection(Connection actualConnection, boolean propagateClose, int maxThreads, int priority) {
        this.actualConnection = actualConnection;
        this.propagateClose = propagateClose;
        executorService = Executors.newFixedThreadPool(maxThreads, new DaemonThreadFactory(priority));
        addShutdownHook();
    }

    private static int getMaxThreads(Dsn dsn) {
        if (dsn.getOptions().containsKey(MAX_THREADS_OPTION)) {
            return Integer.parseInt(dsn.getOptions().get(MAX_THREADS_OPTION));
        } else {
            return DEFAULT_MAX_THREADS;
        }
    }

    private static int getPriority(Dsn dsn) {
        if (dsn.getOptions().containsKey(PRIORITY_OPTION)) {
            return Integer.parseInt(dsn.getOptions().get(PRIORITY_OPTION));
        } else {
            return DEFAULT_PRIORITY;
        }
    }

    private void addShutdownHook() {
        // JUL loggers are shutdown by an other shutdown hook, it's possible that nothing will get actually logged.
        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                try {
                    close();
                } catch (IOException e) {
                    logger.log(Level.SEVERE, "An exception occurred while closing the connection.", e);
                }
            }
        });
    }

    @Override
    public void send(Event event) {
        executorService.execute(new EventSubmitter(event));
    }

    /**
     * {@inheritDoc}.
     * <p><commit_msg>Add documentation to AsyncConnection<commit_after>package net.kencochrane.raven.connection;

import net.kencochrane.raven.Dsn;
import net.kencochrane.raven.event.Event;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Asynchronous usage of a connection.
 * <p>
 * Instead of synchronously sending each event to a connection, use a ThreadPool to establish the connection
 * and submit the event.
 * </p>
 */
public class AsyncConnection implements Connection {
    /**
     * DSN option for the number of threads assigned for the connection.
     */
    public static final String MAX_THREADS_OPTION = "raven.async.threads";
    /**
     * DSN option for the priority of threads assigned for the connection.
     */
    public static final String PRIORITY_OPTION = "raven.async.priority";
    private static final Logger logger = Logger.getLogger(AsyncConnection.class.getCanonicalName());
    /**
     * Timeout of the {@link #executorService}.
     */
    private static final int TIMEOUT = 1000;
    /**
     * Number of threads dedicated to the connection usage by default (Number of processors available).
     */
    private static final int DEFAULT_MAX_THREADS = Runtime.getRuntime().availableProcessors();
    /**
     * Default threads priority.
     */
    private static final int DEFAULT_PRIORITY = Thread.MIN_PRIORITY;
    /**
     * Connection used to actually send the events.
     */
    private final Connection actualConnection;
    /**
     * Executor service in charge of running the connection in separate threads.
     */
    private final ExecutorService executorService;
    /**
     * Option to disable the propagation of the {@link #close()} operation to the actual connection.
     */
    private final boolean propagateClose;

    /**
     * Creates a connection which will rely on an executor to send events.
     * <p>
     * Will propagate the {@link #close()} operation and use {@link #DEFAULT_MAX_THREADS} and {@link #DEFAULT_PRIORITY}.
     * </p>
     *
     * @param actualConnection connection used to send the events.
     */
    public AsyncConnection(Connection actualConnection) {
        this(actualConnection, true, DEFAULT_MAX_THREADS, DEFAULT_PRIORITY);
    }

    /**
     * Creates a connection which will rely on an executor to send events.
     * <p>
     * Will propagate the {@link #close()} operation and attempt to get the number of Threads and their priority from
     * the DSN configuration.
     * </p>
     *
     * @param actualConnection connection used to send the events.
     * @param dsn              Data Source Name containing the additional settings for the async connection.
     */
    public AsyncConnection(Connection actualConnection, Dsn dsn) {
        this(actualConnection, true, getMaxThreads(dsn), getPriority(dsn));
    }

    /**
     * Creates a connection which will rely on an executor to send events.
     *
     * @param actualConnection connection used to send the events.
     * @param propagateClose   whether or not the {@link #actualConnection} should be closed
     *                         when this connection closes.
     * @param maxThreads       number of {@code Thread}s available in the thread pool.
     * @param priority         priority of the {@code Thread}s.
     */
    public AsyncConnection(Connection actualConnection, boolean propagateClose, int maxThreads, int priority) {
        this.actualConnection = actualConnection;
        this.propagateClose = propagateClose;
        executorService = Executors.newFixedThreadPool(maxThreads, new DaemonThreadFactory(priority));
        addShutdownHook();
    }

    /**
     * Gets the number of {@code Thread}s that should be available in the pool.
     * <p>
     * Attempts to get the {@link #MAX_THREADS_OPTION} option from the {@code Dsn},
     * defaults to {@link #DEFAULT_MAX_THREADS} if not available.
     * </p>
     *
     * @param dsn Data Source Name potentially containing settings for the {@link AsyncConnection}.
     * @return the number of threads that should be available in the pool.
     */
    private static int getMaxThreads(Dsn dsn) {
        if (dsn.getOptions().containsKey(MAX_THREADS_OPTION)) {
            return Integer.parseInt(dsn.getOptions().get(MAX_THREADS_OPTION));
        } else {
            return DEFAULT_MAX_THREADS;
        }
    }

    /**
     * Gets the priority of {@code Thread}s in the pool.
     * <p>
     * Attempts to get the {@link #PRIORITY_OPTION} option from the {@code Dsn},
     * defaults to {@link #DEFAULT_PRIORITY} if not available.
     * </p>
     *
     * @param dsn Data Source Name potentially containing settings for the {@link AsyncConnection}.
     * @return the priority of threads available in the pool.
     */
    private static int getPriority(Dsn dsn) {
        if (dsn.getOptions().containsKey(PRIORITY_OPTION)) {
            return Integer.parseInt(dsn.getOptions().get(PRIORITY_OPTION));
        } else {
            return DEFAULT_PRIORITY;
        }
    }

    /**
     * Adds a hook to shutdown the {@link #executorService} gracefully when the JVM shuts down.
     */
    private void addShutdownHook() {
        // JUL loggers are shutdown by an other shutdown hook, it's possible that nothing will get actually logged.
        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                try {
                    close();
                } catch (IOException e) {
                    logger.log(Level.SEVERE, "An exception occurred while closing the connection.", e);
                }
            }
        });
    }

    /**
     * {@inheritDoc}
     * <p>
     * The event will be added to a queue and will be handled by a separate {@code Thread} later on.
     * </p>
     */
    @Override
    public void send(Event event) {
        executorService.execute(new EventSubmitter(event));
    }

    /**
     * {@inheritDoc}.
     * <p>
{'fraction_non_alphanumeric': 0.09053069719042664, 'fraction_numerical': 0.0007567874373285403, 'mean_word_length': 2.7016806722689077, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7489506', 'n_tokens_mistral': 2828, 'n_tokens_neox': 2775, 'n_words': 921}