public class InternalWorker<T> implements Consumer<T>, AutoCloseable {

    private static final Logger LOGGER = LoggerFactory.getLogger(InternalWorker.class);
    private static final String DEFAULT_METRICS_PREFIX = "io.smartcat.berserker";
    private static final String DROPPED = "dropped";
    private static final String WAIT_TIME = "waitTime";
    private static final String SUCCESS_SERVICE_TIME = "successServiceTime";
    private static final String FAILURE_SERVICE_TIME = "failureServiceTime";
    private static final String TOTAL_SERVICE_TIME = "totalServiceTime";
    private static final String SUCCESS_RESPONSE_TIME = "successResponseTime";
    private static final String FAILURE_RESPONSE_TIME = "failureResponseTime";
    private static final String TOTAL_RESPONSE_TIME = "totalResponseTime";
    private static final String GENERATED_THROUGHPUT = "generatedThroughput";
    private static final String SUCCESS_PROCESSED_THROUGHPUT = "successProcessedThroughput";
    private static final String FAILURE_PROCESSED_THROUGHPUT = "failureProcessedThroughput";
    private static final String TOTAL_PROCESSED_THROUGHPUT = "totalProcessedThroughput";
    private static final String QUEUE_SIZE = "queueSize";

    private final LinkedEvictingBlockingQueue<WorkerMeta> queue;
    private final ThreadPoolExecutor threadPoolExecutor;
    private final MetricRegistry metricRegistry;
    private final Meter droppedMeter;
    private final Histogram waitTime;
    private final Histogram successServiceTime;
    private final Histogram failureServiceTime;
    private final Histogram totalServiceTime;
    private final Histogram successResponseTime;
    private final Histogram failureResponseTime;
    private final Histogram totalResponseTime;
    private final Meter generatedThroughput;
    private final Meter totalProcessedThroughput;
    private final Meter successProcessedThroughput;
    private final Meter failureProcessedThroughput;

    private boolean closed = false;

    /**
     * Constructs asynchronous worker with specified <code>delegate</code> worker, <code>queueCapacity</code>.
     * <code>dropFromHead</code> is set to true, <code>metricsPrefix</code> is set to
     * <code>io.smartcat.berserker</code>, <code>threadCount</code> is set to
     * <code>Runtime.getRuntime().availableProcessors()</code>, <code>threadFactory</code> is set to
     * {@link DefaultThreadFactory}.
     *
     * @param delegate Worker which is run in thread pool and to which work is delegated.
     * @param queueCapacity Capacity of the queue used as a message buffer, must be positive number.
     */
    public InternalWorker(Worker<T> delegate, int queueCapacity) {
        this(delegate, queueCapacity, true);
    }

    /**
     * Constructs asynchronous worker with specified <code>delegate</code> worker, <code>queueCapacity</code> and
     * <code>dropFromHead</code>. <code>metricsPrefix</code> is set to <code>io.smartcat.berserker</code>,
     * <code>threadCount</code> is set to <code>Runtime.getRuntime().availableProcessors()</code>,
     * <code>threadFactory</code> is set to {@link DefaultThreadFactory}.
     *
     * @param delegate Worker which is run in thread pool and to which work is delegated.
     * @param queueCapacity Capacity of the queue used as a message buffer, must be positive number.
     * @param dropFromHead If true, message from head of the queue will be dropped, if false, message from tail of the
     *            queue will be dropped.
     */
    public InternalWorker(Worker<T> delegate, int queueCapacity, boolean dropFromHead) {
        this(delegate, queueCapacity, dropFromHead, DEFAULT_METRICS_PREFIX);
    }

    /**
     * Constructs asynchronous worker with specified <code>delegate</code> worker, <code>queueCapacity</code>,
     * <code>dropFromHead</code> and <code>metricsPrefix</code>. <code>threadCount</code> is set to
     * <code>Runtime.getRuntime().availableProcessors()</code>, <code>threadFactory</code> is set to
     * {@link DefaultThreadFactory}.
     *
     * @param delegate Worker which is run in thread pool and to which work is delegated.
     * @param queueCapacity Capacity of the queue used as a message buffer, must be positive number.
     * @param dropFromHead If true, message from head of the queue will be dropped, if false, message from tail of the
     *            queue will be dropped.
     * @param metricsPrefix Prefix for metrics.
     */
    public InternalWorker(Worker<T> delegate, int queueCapacity, boolean dropFromHead, String metricsPrefix) {
        this(delegate, queueCapacity, dropFromHead, metricsPrefix, Runtime.getRuntime().availableProcessors());
    }

    /**
     * Constructs asynchronous worker with specified <code>delegate</code> worker, <code>queueCapacity</code>,
     * <code>dropFromHead</code>, <code>metricsPrefix</code> and <code>threadCount</code>. <code>threadFactory</code> is
     * set to {@link DefaultThreadFactory}.
     *
     * @param delegate Worker which is run in thread pool and to which work is delegated.
     * @param queueCapacity Capacity of the queue used as a message buffer, must be positive number.
     * @param dropFromHead If true, message from head of the queue will be dropped, if false, message from tail of the
     *            queue will be dropped.
     * @param metricsPrefix Prefix for metrics.
     * @param threadCount Number of thread to be used by thread pool, must be positive number.
     */
    public InternalWorker(Worker<T> delegate, int queueCapacity, boolean dropFromHead, String metricsPrefix,
                          int threadCount) {
        this(delegate, queueCapacity, dropFromHead, metricsPrefix, threadCount, new DefaultThreadFactory());
    }

    /**
     * Constructs asynchronous worker with specified <code>delegate</code> worker, <code>queueCapacity</code>,
     * <code>dropFromHead</code>, <code>metricsPrefix</code>, <code>threadCount</code> and <code>threadFactory</code>.
     *
     * @param delegate Worker which is run in thread pool and to which work is delegated.
     * @param queueCapacity Capacity of the queue used as a message buffer, must be positive number.
     * @param dropFromHead If true, message from head of the queue will be dropped, if false, message from tail of the
     *            queue will be dropped.
     * @param metricsPrefix Prefix for metrics.
     * @param threadCount Number of thread to be used by thread pool, must be positive number.
     * @param threadFactory ThreadFactory to be used in creating threads for thread pool.
     */
    public InternalWorker(Worker<T> delegate, int queueCapacity, boolean dropFromHead, String metricsPrefix,
                          int threadCount, ThreadFactory threadFactory) {
        if (delegate == null) {
            throw new IllegalArgumentException("Delegate cannot be null.");
        }
        if (queueCapacity <= 0) {
            throw new IllegalArgumentException("Queue capacity must be positive number.");
        }
        if (threadCount <= 0) {
            throw new IllegalArgumentException("Thread count must be positive number.");
        }
        if (threadFactory == null) {
            throw new IllegalArgumentException("Thread factory cannot be null.");
        }
        this.queue = new LinkedEvictingBlockingQueue<>(dropFromHead, queueCapacity);
        this.threadPoolExecutor = createAndInitThreadPoolExecutor(delegate, threadCount, threadFactory);
        this.metricRegistry = new MetricRegistry();
        this.droppedMeter = metricRegistry.meter(name(metricsPrefix, DROPPED));
        this.waitTime = metricRegistry.histogram(name(metricsPrefix, WAIT_TIME));
        this.successServiceTime = metricRegistry.histogram(name(metricsPrefix, SUCCESS_SERVICE_TIME));
        this.failureServiceTime = metricRegistry.histogram(name(metricsPrefix, FAILURE_SERVICE_TIME));
        this.totalServiceTime = metricRegistry.histogram(name(metricsPrefix, TOTAL_SERVICE_TIME));
        this.successResponseTime = metricRegistry.histogram(name(metricsPrefix, SUCCESS_RESPONSE_TIME));
        this.failureResponseTime = metricRegistry.histogram(name(metricsPrefix, FAILURE_RESPONSE_TIME));
        this.totalResponseTime = metricRegistry.histogram(name(metricsPrefix, TOTAL_RESPONSE_TIME));
        this.generatedThroughput = metricRegistry.meter(name(metricsPrefix, GENERATED_THROUGHPUT));
        this.successProcessedThroughput = metricRegistry.meter(name(metricsPrefix, SUCCESS_PROCESSED_THROUGHPUT));
        this.failureProcessedThroughput = metricRegistry.meter(name(metricsPrefix, FAILURE_PROCESSED_THROUGHPUT));
        this.totalProcessedThroughput = metricRegistry.meter(name(metricsPrefix, TOTAL_PROCESSED_THROUGHPUT));
        metricRegistry.gauge(name(metricsPrefix, QUEUE_SIZE), () -> () -> queue.size());
    }

    /**
     * Accepts message of type {@code <T>}.
     *
     * @param message Message to be processed.
     */
    @Override
    public void accept(T message) {
        if (closed) {
            throw new AlreadyClosedException("Worker is already closed.");
        }
        WorkerMeta meta = new WorkerMeta(message);
        WorkerMeta dropped = queue.put(meta);
        if (dropped != null) {
            droppedMeter.mark();
        }
        generatedThroughput.mark();
    }

    @Override
    public void close() throws Exception {
        if (!closed) {
            threadPoolExecutor.shutdownNow();
            closed = true;
        }
    }

    /**
     * Returns metric registry this async worker is using.
     *
     * @return Metric registry this async worker is using.
     */
    public MetricRegistry getMetricRegistry() {
        return metricRegistry;
    }

    private ThreadPoolExecutor createAndInitThreadPoolExecutor(Worker<T> delegate, int threadCount,
            ThreadFactory threadFactory) {
        ThreadPoolExecutor result = new ThreadPoolExecutor(threadCount, threadCount, 10, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(), threadFactory);
        while (result.getPoolSize() < threadCount) {
            result.submit(() -> {
                while (true) {
                    try {
                        WorkerMeta meta = queue.take();
                        meta.markAsAccepted();
                        delegate.accept(meta.getPayload(), () -> {
                            meta.markAsDone();
                            waitTime.update(meta.getWaitNanoTime());
                            successServiceTime.update(meta.getServiceNanoTime());
                            totalServiceTime.update(meta.getServiceNanoTime());
                            successResponseTime.update(meta.getResponseNanoTime());
                            totalResponseTime.update(meta.getResponseNanoTime());
                            successProcessedThroughput.mark();
                            totalProcessedThroughput.mark();
                        }, () -> {
                            meta.markAsDone();
                            waitTime.update(meta.getWaitNanoTime());
                            failureServiceTime.update(meta.getServiceNanoTime());
                            totalServiceTime.update(meta.getServiceNanoTime());
                            failureResponseTime.update(meta.getResponseNanoTime());
                            totalResponseTime.update(meta.getResponseNanoTime());
                            failureProcessedThroughput.mark();
                            totalProcessedThroughput.mark();
                        });
                    } catch (Exception e) {
                        String workerName = delegate.getClass().getName();
                        LOGGER.error("Error while accepting payload at worker: " + workerName + ". Error: ", e);
                    }
                }
            });
        }
        return result;
    }

    private String name(String metricsPrefix, String name) {
        String prefix = metricsPrefix == null || metricsPrefix.isEmpty() ? DEFAULT_METRICS_PREFIX : metricsPrefix;
        return MetricRegistry.name(prefix, name);
    }

    /**
     * Thread factory that creates normal priority, daemon threads with 'async-worker-thread-&lt;number&gt;' names.
     */
    public static class DefaultThreadFactory implements ThreadFactory {

        private final AtomicInteger threadNumber = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("async-worker-thread-" + threadNumber.getAndIncrement());
            thread.setDaemon(true);
            thread.setPriority(Thread.NORM_PRIORITY);
            return thread;
        }
    }

    /**
     * Meta data on worker message processing.
     */
    private class WorkerMeta {

        private final T payload;
        private final long timeSubmittedInNanos;
        private long timeAcceptedInNanos;
        private long timeDoneInNanos;
        private boolean dropped = false;

        WorkerMeta(T payload) {
            this.payload = payload;
            this.timeSubmittedInNanos = now();
        }

        void markAsAccepted() {
            timeAcceptedInNanos = now();
        }

        void markAsDone() {
            timeDoneInNanos = now();
        }

        /**
         * Returns payload sent to worker.
         *
         * @return payload sent to worker.
         */
        T getPayload() {
            return payload;
        }

        /**
         * Returns time in nanoseconds when message was submitted to the worker thread.
         *
         * @return time in nanoseconds when message was submitted to the worker thread.
         */
        long getWaitNanoTime() {
            return timeAcceptedInNanos - timeSubmittedInNanos;
        }

        /**
         * Returns time in nanoseconds when message was accepted by the worker.
         *
         * @return time in nanoseconds when message was accepted by the worker.
         */
        long getServiceNanoTime() {
            return timeDoneInNanos - timeAcceptedInNanos;
        }

        /**
         * Returns time in nanoseconds when processing was done on message.
         *
         * @return time in nanoseconds when processing was done on message.
         */
        long getResponseNanoTime() {
            return timeDoneInNanos - timeSubmittedInNanos;
        }

        @Override
        public String toString() {
            return "WorkerMeta [payload=" + payload + ", timeSubmittedInNanos=" + timeSubmittedInNanos
                    + ", timeAcceptedInNanos=" + timeAcceptedInNanos + ", timeDoneInNanos=" + timeDoneInNanos
                    + ", dropped=" + dropped + "]";
        }

        private long now() {
            return System.nanoTime();
        }
    }
} public class LoadGenerator<T> {

    private static final Logger LOGGER = LoggerFactory.getLogger(LoadGenerator.class);
    private static final long NANOS_IN_SECOND = TimeUnit.SECONDS.toNanos(1);
    private static final long TICK_PERIOD_IN_NANOS = 1000;

    private final DataSource<T> dataSource;
    private final RateGenerator rateGenerator;
    private final Consumer<T> worker;

    private AtomicBoolean terminate = new AtomicBoolean(false);

    /**
     * Constructs load generator with specified <code>dataSource</code>, <code>rateGenerator</code> and
     * <code>worker</code>.
     *
     * @param dataSource Data source from which load generator polls data.
     * @param rateGenerator Rate generator which generates rate based on time point.
     * @param worker Worker which accepts data polled from <code>dataSource</code> at rate provided by
     *            <code>rateGenerator</code>.
     */
    public LoadGenerator(DataSource<T> dataSource, RateGenerator rateGenerator, Consumer<T> worker) {
        this.dataSource = dataSource;
        this.rateGenerator = rateGenerator;
        this.worker = worker;
    }

    /**
     * Runs load generator.
     *
     * @throws IllegalStateException When run is attempted after load generator was terminated.
     */
    public void run() {
        try {
            checkState();
            LOGGER.info("Load generator started.");
            long beginning = System.nanoTime();
            long previous = beginning;
            infiniteWhile: while (true) {
                if (terminate.get()) {
                    LOGGER.info("Termination signal detected. Terminating load generator...");
                    break;
                }
                long now = System.nanoTime();
                long fromBeginning = now - beginning;
                long elapsed = now - previous;
                double rate = rateGenerator.getRate(fromBeginning);
                long normalizedRate = normalizeRate(elapsed, rate);
                if (normalizedRate > 0) {
                    previous += calculateConsumedTime(normalizedRate, rate);
                }
                for (int i = 0; i < normalizedRate; i++) {
                    if (!dataSource.hasNext(fromBeginning)) {
                        LOGGER.info("Reached end of data source. Terminating load generator...");
                        terminate.set(true);
                        break infiniteWhile;
                    }
                    T data = dataSource.getNext(fromBeginning);
                    worker.accept(data);
                }
            }
            LOGGER.info("Load generator terminated.");
        } catch (Exception e) {
            LOGGER.error("Terminating load generator due to error. Error: ", e);
        }
    }

    /**
     * Stops load generator.
     */
    public void terminate() {
        terminate.set(true);
        LOGGER.info("Termination signal sent.");
    }

    private void checkState() {
        if (terminate.get()) {
            throw new IllegalStateException("Load generator is stopped and cannot be started again.");
        }
    }

    private long normalizeRate(long elapsed, double rate) {
        if (elapsed < TICK_PERIOD_IN_NANOS) {
            return 0;
        }
        return (long) (elapsed * rate / NANOS_IN_SECOND);
    }

    private long calculateConsumedTime(long normalizedRate, double rate) {
        return (long) (normalizedRate * NANOS_IN_SECOND / rate);
    }
} public interface DataSource<T> {

    /**
     * Returns true if data source can provide next value, otherwise false.
     *
     * @param time Relative time in nanoseconds from starting load generator. Time can be used for implementing time
     *            dependent data source which can then return data with time stamps and/or data influenced by time in
     *            any other way.
     * @return True if data source can provide next value, otherwise false.
     */
    boolean hasNext(long time);

    /**
     * Returns next value from this data source.
     *
     * @param time Relative time in nanoseconds from starting load generator. Time can be used for implementing time
     *            dependent data source which can then return data with time stamps and/or data influenced by time in
     *            any other way.
     * @return Next value from this data source.
     */
    T getNext(long time);
} public interface RateGenerator {

    /**
     * Returns rate per second as a function of time.
     *
     * @param time Relative time in nanoseconds from starting load generator.
     * @return Rate per second as a function of time.
     */
    double getRate(long time);
} public interface Worker<T> {

    /**
     * Accepts message of type {@code <T>} and processes it.
     *
     * @param message Message which will be processed.
     * @param commitSuccess Callback to be invoked when processing is successful.
     * @param commitFailure Callback to be invoked in case of a failure.
     */
    void accept(T message, Runnable commitSuccess, Runnable commitFailure);
} public interface BaseConfiguration {

    /**
     * Returns name of this configuration.
     *
     * @return name of this configuration.
     */
    String getName();
} public class ConfigurationParseException extends ConfigurationException {

    private static final long serialVersionUID = -7101805420460054579L;

    /**
     * Default constructor.
     */
    public ConfigurationParseException() {
    }

    /**
     * Constructor.
     *
     * @param message Error message.
     * @param cause Exception cause.
     * @param enableSuppression controls exception suppression.
     * @param writableStackTrace stack trace.
     */
    public ConfigurationParseException(String message, Throwable cause, boolean enableSuppression,
            boolean writableStackTrace) {
        super(message, cause, enableSuppression, writableStackTrace);
    }

    /**
     * Constructor.
     *
     * @param message Error message.
     * @param cause Exception cause.
     */
    public ConfigurationParseException(String message, Throwable cause) {
        super(message, cause);
    }

    /**
     * Constructor.
     *
     * @param message Error message.
     */
    public ConfigurationParseException(String message) {
        super(message);
    }

    /**
     * Constructor.
     *
     * @param cause Exception cause.
     */
    public ConfigurationParseException(Throwable cause) {
        super(cause);
    }
} public interface DataSourceConfiguration extends BaseConfiguration {

    /**
     * Returns data source based on configuration parameters.
     *
     * @param configuration Configuration specific to data source it should construct.
     * @return Instance of {@link DataSource}, never null.
     *
     * @throws ConfigurationParseException If there is an error during configuration parsing.
     */
    DataSource<?> getDataSource(Map<String, Object> configuration) throws ConfigurationParseException;
} public class GlobalConfiguration {

    /**
     * Configuration specific to {@link io.smartcat.berserker.api.DataSource DataSource}.
     */
    public Map<String, Object> dataSourceConfiguration;

    /**
     * Configuration specific to {@link io.smartcat.berserker.api.Worker Worker}.
     */
    public Map<String, Object> workerConfiguration;

    /**
     * Configuration specific to {@link io.smartcat.berserker.api.RateGenerator RateGenerator}.
     */
    public Map<String, Object> rateGeneratorConfiguration;

    /**
     * Configuration specific to {@link com.codahale.metrics.Reporter Reporter}.
     */
    public Map<String, Object> metricsReporterConfiguration;

    /**
     * Configuration explaining which {@link DataSourceConfiguration}, {@link WorkerConfiguration} and
     * {@link RateGeneratorConfiguration} implementations will be used.
     */
    public LoadGeneratorConfiguration loadGeneratorConfiguration;
} public class LoadGeneratorConfiguration {

    /**
     * Name of {@link io.smartcat.berserker.api.DataSource DataSource} configuration.
     */
    public String dataSourceConfigurationName;

    /**
     * Name of {@link io.smartcat.berserker.api.RateGenerator RateGenerator} configuration.
     */
    public String rateGeneratorConfigurationName;

    /**
     * Name of {@link io.smartcat.berserker.api.Worker Worker} configuration.
     */
    public String workerConfigurationName;

    /**
     * Name of {@link com.codahale.metrics.Reporter Reporter} configuration.
     */
    public String metricsReporterConfigurationName;

    /**
     * Prefix used for metrics.
     */
    public String metricsPrefix;

    /**
     * Number of threads to be used by Load Generator.
     */
    public int threadCount;

    /**
     * Capacity of queue to be used by Load Generator.
     */
    public int queueCapacity;

    /**
     * Validates this configuration.
     *
     * @throws ConfigurationException If configuration is not correct.
     */
    public void validate() throws ConfigurationException {
        validateProperty(dataSourceConfigurationName, "data-source-configuration-name");
        validateProperty(rateGeneratorConfigurationName, "rate-generator-configuration-name");
        validateProperty(workerConfigurationName, "worker-configuration-name");
        validateProperty(threadCount, "thread-count");
        validateProperty(queueCapacity, "queue-capacity");
    }

    private void validateProperty(int property, String propertyName) throws ConfigurationException {
        if (property <= 0) {
            throw new ConfigurationException(
                    "Property: '" + propertyName + "' is mandatory. It needs to be positive number.");
        }
    }
    private void validateProperty(String property, String propertyName) throws ConfigurationException {
        if (property == null || property.isEmpty()) {
            throw new ConfigurationException(
                    "Property: '" + propertyName + "' is mandatory. It cannot be null nor empty.");
        }
    }
} public interface MetricsReporterConfiguration extends BaseConfiguration {

    /**
     * Creates and starts reporter for specified <code>metricRegistry</code> based on provided configuration.
     *
     * @param metricRegistry Metric registry for which to create and start reporter.
     * @param configuration Configuration specific to metrics reporter it should construct.
     */
    void createAndStartReporter(MetricRegistry metricRegistry, Map<String, Object> configuration);
} public interface RateGeneratorConfiguration extends BaseConfiguration {

    /**
     * Returns rate generator based on configuration parameters.
     *
     * @param configuration Configuration specific to rate generator it should construct.
     * @return Instance of {@link RateGenerator}, never null.
     *
     * @throws ConfigurationParseException If there is an error during configuration parsing.
     */
    RateGenerator getRateGenerator(Map<String, Object> configuration) throws ConfigurationParseException;
} public interface WorkerConfiguration extends BaseConfiguration {

    /**
     * Returns worker based on configuration parameters.
     *
     * @param configuration Configuration specific to this worker.
     * @return Instance of {@link Worker}, never null.
     *
     * @throws ConfigurationParseException If there is an error during configuration parsing.
     */
    Worker<?> getWorker(Map<String, Object> configuration) throws ConfigurationParseException;
} public class YamlConfigurationLoader implements ConfigurationLoader {

    private static final Logger LOGGER = LoggerFactory.getLogger(YamlConfigurationLoader.class);
    private static final String HYPHEN = "-";
    private static final String CONFIGURATION_PROPERTY_NAME = "berserker.config";

    /**
     * Default external configuration file name.
     */
    private static final String DEFAULT_CONFIGURATION_URL = "berserker-default.yml";

    /**
     * Determines and returns the external configuration URL.
     *
     * @return {@link URL} configuration file URL
     * @throws ConfigurationException in case of a bogus URL
     */
    private URL getStorageConfigUrl() throws ConfigurationException {
        String configUrl = System.getProperty(CONFIGURATION_PROPERTY_NAME);
        if (configUrl == null) {
            configUrl = DEFAULT_CONFIGURATION_URL;
            LOGGER.info("Using default configuration " + DEFAULT_CONFIGURATION_URL);
        }

        URL url;
        try {
            url = new URL(configUrl);
            url.openStream().close(); // catches well-formed but bogus URLs
        } catch (Exception err) {
            ClassLoader loader = YamlConfigurationLoader.class.getClassLoader();
            url = loader.getResource(configUrl);
            if (url == null) {
                String required = "file:" + File.separator + File.separator;
                if (!configUrl.startsWith(required)) {
                    throw new ConfigurationException("Expecting URI in variable [" + CONFIGURATION_PROPERTY_NAME + "]. "
                            + "Please prefix the file with " + required + File.separator + " for local files or "
                            + required + "<server>" + File.separator + " for remote files. Aborting.");
                }
                throw new ConfigurationException(
                        "Cannot locate " + configUrl + ".  If this is a local file, please confirm you've provided "
                                + required + File.separator + " as a URI prefix.");
            }
        }
        return url;
    }

    @Override
    public GlobalConfiguration loadConfig() throws ConfigurationException {
        return loadConfig(getStorageConfigUrl());
    }

    @Override
    public GlobalConfiguration loadConfig(URL url) throws ConfigurationException {
        try {
            LOGGER.info("Loading settings from {}", url);
            Constructor constructor = new Constructor(GlobalConfiguration.class);

            // treat dashed properties as camel case properties
            constructor.setPropertyUtils(new PropertyUtils() {
                @Override
                public Property getProperty(Class<? extends Object> type, String name) throws IntrospectionException {
                    if (name.contains(HYPHEN)) {
                        name = LOWER_HYPHEN.to(LOWER_CAMEL, name);
                    }
                    return super.getProperty(type, name);
                }
            });

            Yaml yaml = new Yaml(constructor);
            GlobalConfiguration result;
            try (InputStream is = url.openStream()) {
                result = yaml.loadAs(is, GlobalConfiguration.class);
            } catch (IOException e) {
                throw new AssertionError(e);
            }

            if (result == null) {
                throw new ConfigurationException("Invalid yaml");
            }
            return result;
        } catch (YAMLException e) {
            throw new ConfigurationException("Invalid yaml", e);
        }
    }
}
import java.io.File;
import java.io.IOException;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import io.smartcat.berserker.worker.InternalWorker;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.reflections.Reflections;
import com.codahale.metrics.MetricRegistry;
import io.smartcat.berserker.LoadGenerator;
import io.smartcat.berserker.api.DataSource;
import io.smartcat.berserker.api.RateGenerator;
import io.smartcat.berserker.api.Worker;
import io.smartcat.berserker.configuration.BaseConfiguration;
import io.smartcat.berserker.configuration.ConfigurationParseException;
import io.smartcat.berserker.configuration.DataSourceConfiguration;
import io.smartcat.berserker.configuration.GlobalConfiguration;
import io.smartcat.berserker.configuration.LoadGeneratorConfiguration;
import io.smartcat.berserker.configuration.MetricsReporterConfiguration;
import io.smartcat.berserker.configuration.RateGeneratorConfiguration;
import io.smartcat.berserker.configuration.WorkerConfiguration;
import io.smartcat.berserker.configuration.YamlConfigurationLoader;
package io.smartcat.berserker.runner;





/**
 * Runner which takes configuration file, constructs {@link LoadGenerator} with depending {@link DataSource},
 * {@link RateGenerator} and {@link Worker} and runs load generator.
 */
public class LoadGeneratorRunner {

    private static final String BERSERKER_BASE_PACKAGE = "io.smartcat.berserker";
    private static final String CONFIG_SHORT = "c";
    private static final String CONFIG_LONG = "config";

    private LoadGeneratorRunner() {
    }

    /**
     * Main method for starting load generator runner.
     *
     * @param args Command line arguments.
     */
    public static void main(String[] args) {
        Options options = getOptions();
        CommandLineParser parser = new DefaultParser();
        CommandLine cmd = null;
        try {
            cmd = parser.parse(options, args);
        } catch (ParseException e) {
            HelpFormatter formatter = new HelpFormatter();
            formatter.printHelp("java -jar berserker-runner.jar -c <path_to_config_file>", options);
        }
        if (cmd == null) {
            return;
        }
        try {
            generateLoad(cmd.getOptionValue(CONFIG_LONG));
        } catch (Exception e) {
            throw e instanceof RuntimeException ? (RuntimeException) e : new RuntimeException(e);
        }
    }

    @SuppressWarnings({ "unchecked", "rawtypes" })
    private static void generateLoad(String configFilePath) throws Exception {
        YamlConfigurationLoader configurationLoader = new YamlConfigurationLoader();
        GlobalConfiguration configuration = configurationLoader.loadConfig(getURL(configFilePath));
        LoadGeneratorConfiguration loadGeneratorConfiguration = configuration.loadGeneratorConfiguration;
        loadGeneratorConfiguration.validate();
        DataSource dataSource = getDataSource(loadGeneratorConfiguration.dataSourceConfigurationName,
                configuration.dataSourceConfiguration);
        RateGenerator rateGenerator = getRateGenerator(loadGeneratorConfiguration.rateGeneratorConfigurationName,
                configuration.rateGeneratorConfiguration);
        Worker workerDelegate = getWorker(loadGeneratorConfiguration.workerConfigurationName,