public class Address implements Iterable<Address.Tuple> {

    public static Address apply(String address) {
        List<String> tokens = address == null ? Collections.<String>emptyList() :
                Splitter.on(CharMatcher.anyOf("/="))
                        .trimResults()
                        .omitEmptyStrings()
                        .splitToList(address);

        List<Tuple> tuples = new ArrayList<>(tokens.size() / 2 + 1);
        for (Iterator<String> iterator = tokens.iterator(); iterator.hasNext(); ) {
            String type = iterator.next();
            String name = iterator.hasNext() ? iterator.next() : "";
            tuples.add(new Tuple(type, name));
        }

        return new Address(tuples);
    }


    private final List<Tuple> tuples;

    private Address(final List<Tuple> tuples) {
        this.tuples = new ArrayList<>();
        if (tuples != null) {
            this.tuples.addAll(tuples);
        }
    }

    @Override
    public boolean equals(final Object o) {
        if (this == o) { return true; }
        if (!(o instanceof Address)) { return false; }

        Address address = (Address) o;

        if (!tuples.equals(address.tuples)) { return false; }

        return true;
    }

    @Override
    public int hashCode() {
        return tuples.hashCode();
    }

    @Override
    public String toString() {
        return Joiner.on('/').join(tuples);
    }

    @Override
    public Iterator<Tuple> iterator() {
        return Iterators.unmodifiableIterator(tuples.iterator());
    }

    public boolean isEmpty() {return tuples.isEmpty();}

    public boolean isBalanced() {
        for (Tuple tuple : this) {
            if (tuple.getValue() == null || tuple.getValue().length() == 0) {
                return false;
            }
        }
        return true;
    }

    public boolean startsWith(Tuple tuple) {
        return !tuples.isEmpty() && tuples.get(0).equals(tuple);
    }


    /**
     * @author Harald Pehl
     */
    public static class Tuple {

        public static Tuple apply(String tuple) {
            if (tuple == null) {
                throw new IllegalArgumentException("Tuple must not be null");
            }
            List<String> tuples = Splitter.on('=')
                    .omitEmptyStrings()
                    .trimResults()
                    .splitToList(tuple);
            if (tuples.isEmpty() || tuples.size() != 2) {
                throw new IllegalArgumentException("Malformed tuple: " + tuple);
            }
            return new Tuple(tuples.get(0), tuples.get(1));
        }

        private final String key;
        private final String value;

        private Tuple(final String key, final String value) {
            this.key = key;
            this.value = value;
        }

        @Override
        public boolean equals(final Object o) {
            if (this == o) { return true; }
            if (!(o instanceof Tuple)) { return false; }

            Tuple that = (Tuple) o;

            if (!value.equals(that.value)) { return false; }
            if (!key.equals(that.key)) { return false; }

            return true;
        }

        @Override
        public int hashCode() {
            int result = key.hashCode();
            result = 31 * result + value.hashCode();
            return result;
        }

        @Override
        public String toString() {
            return key + "=" + value;
        }

        public String getKey() {
            return key;
        }

        public String getValue() {
            return value;
        }
    }
} public interface Configuration {

    public enum Diagnostics {STORAGE, CONSOLE};
    public enum Storage {RHQ, INFLUX}

    /**
     * The host controller host.
     * @return
     */
    String getHost();

    /**
     * The host controller port.
     *
     * @return
     */
    int getPort();

    /**
     * Host controller user
     * @return
     */
    String getUsername();

    /**
     * Host controller password
     * @return
     */
    String getPassword();

    /**
     * Number of threads the scheduler uses to poll for new data.
     *
     * @return
     */
    int getSchedulerThreads();

    /**
     * The resources that are to be monitored.
     * {@link org.wildfly.metrics.scheduler.config.ResourceRef}'s use relative addresses.
     * The core {@link org.wildfly.metrics.scheduler.Service} will resolve it against absolute address within a Wildfly domain.
     *
     * @return
     */
    List<ResourceRef> getResourceRefs();

    Storage getStorageAdapter();

    String getStorageUrl();

    String getStorageUser();

    String getStoragePassword();

    String getStorageDBName();

    String getStorageToken();

    Diagnostics getDiagnostics();

} public class ResourceRef {

    private final String address;
    private final String attribute;
    private final Interval interval;

    public ResourceRef(final String address, final String attribute, final Interval interval) {
        this.address = address;
        this.attribute = attribute;
        this.interval = interval;
    }

    @Override
    public String toString() {
        return "Task(" + address + ":" + attribute + ", " + interval + ")";
    }

    public String getAddress() {
        return address;
    }

    public String getAttribute() {
        return attribute;
    }

    public Interval getInterval() {
        return interval;
    }
} public interface Diagnostics {
    Timer getRequestTimer();
    Meter getErrorRate();
    Meter getDelayedRate();

    Meter getStorageErrorRate();
    Counter getStorageBufferSize();
} public class StorageReporter extends ScheduledReporter {

    private static StorageAdapter storageAdapter;
    private final Locale locale;
    private final Clock clock;
    private final DateFormat dateFormat;

    private StorageReporter(MetricRegistry registry,
                            Locale locale,
                            Clock clock,
                            TimeZone timeZone,
                            TimeUnit rateUnit,
                            TimeUnit durationUnit,
                            MetricFilter filter, StorageAdapter storageAdapter) {
        super(registry, "storage-reporter", filter, rateUnit, durationUnit);
        this.locale = locale;
        this.clock = clock;
        this.storageAdapter = storageAdapter;
        this.dateFormat = DateFormat.getDateTimeInstance(DateFormat.SHORT,
                DateFormat.MEDIUM,
                locale);
        dateFormat.setTimeZone(timeZone);
    }


    @Override
    public void report(
            SortedMap<String, Gauge> gauges,
            SortedMap<String, Counter> counters,
            SortedMap<String, Histogram> histograms,
            SortedMap<String, Meter> meters,
            SortedMap<String, Timer> timers) {

        if (!gauges.isEmpty()) {

            Set<DataPoint> samples = new HashSet<>(gauges.size());
            for (Map.Entry<String, Gauge> entry : gauges.entrySet()) {
                Gauge<Integer> gauge = entry.getValue();

                samples.add(
                        new DataPoint(
                                new Task("foo", "bar", Address.apply("service=metric-scheduler"), entry.getKey(), null, Interval.EACH_SECOND),
                                gauge.getValue()
                        )
                );
            }
            storageAdapter.store(samples);
        }

        if (!counters.isEmpty()) {

            Set<DataPoint> samples = new HashSet<>(counters.size());
            for (Map.Entry<String, Counter> entry : counters.entrySet()) {
                samples.add(
                        new DataPoint(
                                new Task("foo", "bar", Address.apply("service=metric-scheduler"), entry.getKey(), null, Interval.EACH_SECOND),
                                entry.getValue().getCount()
                        )
                );
            }

            storageAdapter.store(samples);

        }

        /*if (!histograms.isEmpty()) {

            Set<Sample> samples = new HashSet<>(histograms.size());
            for (Map.Entry<String, Histogram> entry : histograms.entrySet()) {
                Histogram hist = entry.getValue();
                samples.add(
                        new Sample(
                                new Task(Address.apply("service=metric-scheduler"), entry.getKey(), null, Interval.EACH_SECOND),
                                0 // TODO
                        )
                );
            }
            storageAdapter.store(samples);
        }*/

        if (!meters.isEmpty()) {

            Set<DataPoint> samples = new HashSet<>(meters.size());
            for (Map.Entry<String, Meter> entry : meters.entrySet()) {
                Meter meter = entry.getValue();
                samples.add(
                        new DataPoint(
                                new Task("foo", "bar", Address.apply("service=metric-scheduler"), entry.getKey(), null, Interval.EACH_SECOND),
                                meter.getOneMinuteRate()
                        )
                );
            }
            storageAdapter.store(samples);
        }

        if (!timers.isEmpty()) {

            Set<DataPoint> samples = new HashSet<>(timers.size());

            for (Map.Entry<String, Timer> entry : timers.entrySet()) {
                Timer timer = entry.getValue();

                samples.add(
                        new DataPoint(
                                new Task("foo", "bar", Address.apply("service=metric-scheduler"), entry.getKey(), null, Interval.EACH_SECOND),
                                timer.getSnapshot().get75thPercentile()
                        )
                );
            }

            storageAdapter.store(samples);
        }

    }

    public static Builder forRegistry(MetricRegistry registry, StorageAdapter storageAdapter) {
        return new Builder(registry, storageAdapter);
    }

    public static class Builder {
        private final MetricRegistry registry;
        private final StorageAdapter storageAdapter;
        private PrintStream output;
        private Locale locale;
        private Clock clock;
        private TimeZone timeZone;
        private TimeUnit rateUnit;
        private TimeUnit durationUnit;
        private MetricFilter filter;

        private Builder(MetricRegistry registry, StorageAdapter storageAdapter) {
            this.registry = registry;
            this.storageAdapter = storageAdapter;
            this.locale = Locale.getDefault();
            this.clock = Clock.defaultClock();
            this.timeZone = TimeZone.getDefault();
            this.rateUnit = TimeUnit.SECONDS;
            this.durationUnit = TimeUnit.MILLISECONDS;
            this.filter = MetricFilter.ALL;
        }

        /**
         * Write to the given {@link PrintStream}.
         *
         * @param output a {@link PrintStream} instance.
         * @return {@code this}
         */
        public Builder outputTo(PrintStream output) {
            this.output = output;
            return this;
        }

        /**
         * Format numbers for the given {@link Locale}.
         *
         * @param locale a {@link Locale}
         * @return {@code this}
         */
        public Builder formattedFor(Locale locale) {
            this.locale = locale;
            return this;
        }

        /**
         * Use the given {@link Clock} instance for the time.
         *
         * @param clock a {@link Clock} instance
         * @return {@code this}
         */
        public Builder withClock(Clock clock) {
            this.clock = clock;
            return this;
        }

        /**
         * Use the given {@link TimeZone} for the time.
         *
         * @param timeZone a {@link TimeZone}
         * @return {@code this}
         */
        public Builder formattedFor(TimeZone timeZone) {
            this.timeZone = timeZone;
            return this;
        }

        /**
         * Convert rates to the given time unit.
         *
         * @param rateUnit a unit of time
         * @return {@code this}
         */
        public Builder convertRatesTo(TimeUnit rateUnit) {
            this.rateUnit = rateUnit;
            return this;
        }

        /**
         * Convert durations to the given time unit.
         *
         * @param durationUnit a unit of time
         * @return {@code this}
         */
        public Builder convertDurationsTo(TimeUnit durationUnit) {
            this.durationUnit = durationUnit;
            return this;
        }

        /**
         * Only report metrics which match the given filter.
         *
         * @param filter a {@link MetricFilter}
         * @return {@code this}
         */
        public Builder filter(MetricFilter filter) {
            this.filter = filter;
            return this;
        }

        public StorageReporter build() {
            return new StorageReporter(registry,
                    locale,
                    clock,
                    timeZone,
                    rateUnit,
                    durationUnit,
                    filter, storageAdapter
            );
        }
    }

} public class ClientFactoryImpl implements ModelControllerClientFactory  {

    private String host;
    private int port;
    private String username;
    private String password;

    public ClientFactoryImpl(String host, int port, String username, String password) {
        this.host = host;
        this.port = port;
        this.username = username;
        this.password = password;
    }

    @Override
    public ModelControllerClient createClient() {

        final CallbackHandler callbackHandler = new CallbackHandler() {

            public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
                for (Callback current : callbacks) {
                    if (current instanceof NameCallback) {
                        NameCallback ncb = (NameCallback) current;
                        ncb.setName(username);
                    } else if (current instanceof PasswordCallback) {
                        PasswordCallback pcb = (PasswordCallback) current;
                        pcb.setPassword(password.toCharArray());
                    } else if (current instanceof RealmCallback) {
                        RealmCallback rcb = (RealmCallback) current;
                        rcb.setText(rcb.getDefaultText());
                    } else {
                        throw new UnsupportedCallbackException(current);
                    }
                }
            }
        };

        try {
            return ModelControllerClient.Factory.create(InetAddress.getByName(host), port, callbackHandler);
        } catch (UnknownHostException e) {
            throw new RuntimeException("Failed ot create controller client", e);
        }
    }
} public class IntervalBasedScheduler extends AbstractScheduler {

    private final ScheduledExecutorService executorService;
    private final List<ScheduledFuture> jobs;
    private final int poolSize;

    private final ModelControllerClientFactory clientFactory;
    private final Diagnostics monitor;

    private ConcurrentLinkedQueue<ModelControllerClient> connectionPool = new ConcurrentLinkedQueue<>();

    public IntervalBasedScheduler(ModelControllerClientFactory clientFactory, Diagnostics monitor, final int poolSize) {
        this.clientFactory = clientFactory;
        this.monitor = monitor;
        this.poolSize = poolSize;

        this.executorService = Executors.newScheduledThreadPool(poolSize, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                SchedulerLogger.LOGGER.debug("Creating new executor thread");
                return new Thread(r);
            }
        });

        this.jobs = new LinkedList<>();

    }

    @Override
    public void schedule(List<Task> tasks, final CompletionHandler completionHandler) {
        verifyState(STOPPED);

        // optimize task groups
        List<TaskGroup> groups = new IntervalGrouping().apply(tasks);

        SchedulerLogger.LOGGER.infof("Number of tasks: %s", tasks.size());
        SchedulerLogger.LOGGER.infof("Number of task groups: %s", groups.size());

        // populate connection pool
        for (int i = 0; i < poolSize; i++) {
            try {
                connectionPool.add(
                        clientFactory.createClient()
                );
            } catch (Throwable e) {
                e.printStackTrace();
            }
        }

        // schedule IO
        for (TaskGroup group : groups) {
            jobs.add(

                    // schedule tasks
                    executorService.scheduleWithFixedDelay(
                            new IO(group, completionHandler),
                            group.getOffsetMillis(), group.getInterval().millis(),
                            MILLISECONDS
                    )
            );
        }

        pushState(RUNNING);
    }

    @Override
    public void shutdown() {
        verifyState(RUNNING);


        try {
            for (ScheduledFuture job : jobs) {
                job.cancel(false);
            }
            executorService.shutdown();
            executorService.awaitTermination(5, TimeUnit.SECONDS);

        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {

            // cleanup pool
            for (ModelControllerClient client : connectionPool) {
                try {
                    client.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            };

            pushState(STOPPED);
        }
    }

    private class IO implements Runnable {

        private static final String OUTCOME = "outcome";
        private static final String RESULT = "result";
        private static final String FAILURE_DESCRIPTION = "failure-description";
        private static final String SUCCESS = "success";

        private final TaskGroup group;
        private final CompletionHandler completionHandler;
        private final ModelNode operation;

        private IO(TaskGroup group, CompletionHandler completionHandler) {
            this.group = group;
            this.completionHandler = completionHandler;

            // for the IO lifetime the operation is immutable and can be re-used
            this.operation = new ReadAttributeOperationBuilder().createOperation(group);
        }

        @Override
        public void run() {

            if(connectionPool.isEmpty())
                throw new IllegalStateException("Connection pool expired!");
            final ModelControllerClient client = connectionPool.poll();

            try {

                Timer.Context requestContext = monitor.getRequestTimer().time();

                // execute request
                ModelNode response = client.execute(operation);

                long durationMs = requestContext.stop() / 1000000;

                String outcome = response.get(OUTCOME).asString();
                if (SUCCESS.equals(outcome))
                {

                    if (durationMs > group.getInterval().millis()) {
                        monitor.getDelayedRate().mark(1);
                    }

                    List<Property> steps = response.get(RESULT).asPropertyList();
                    assert steps.size() == group.size() : "group structure doesn't match actual response structure";

                    int i=0;
                    for (Property step : steps) {
                        Task task = group.getTask(i);

                        // deconstruct model node
                        ModelNode data = step.getValue();
                        Double value = null;
                        if(task.getSubref()!=null)
                        {
                            value = data.get(RESULT).get(task.getSubref()).asDouble();
                        }
                        else
                        {
                            value = data.get(RESULT).asDouble();
                        }

                        completionHandler.onCompleted(new DataPoint(task, value));
                        i++;
                    }


                } else {
                    monitor.getErrorRate().mark(1);
                    completionHandler.onFailed(new RuntimeException(response.get(FAILURE_DESCRIPTION).asString()));
                }

            } catch (Throwable e) {
                monitor.getErrorRate().mark(1);
                completionHandler.onFailed(e);
            } finally {

                // return to pool
                connectionPool.add(client);

            }
        }

    }

} public interface Scheduler {

    public enum State {RUNNING, STOPPED}

    void schedule(List<Task> operations, CompletionHandler completionHandler);

    void shutdown();

    /**
     * Callback for completed tasks
     */
    interface CompletionHandler {
        void onCompleted(DataPoint sample);
        void onFailed(Throwable e);
    }
} public class Task {

    private final String host;
    private final String server;
    private final Address address;
    private final String attribute;
    private final String subref;
    private final Interval interval;

    public Task(
            String host, String server,
            Address address,
            String attribute,
            String subref,
            Interval interval
    ) {
        this.host = host;
        this.server = server;
        this.address = address;
        this.attribute = attribute;
        this.subref = subref;
        this.interval = interval;
    }

    public Address getAddress() {
        return address;
    }

    public String getAttribute() {
        return attribute;
    }

    public Interval getInterval() {
        return interval;
    }

    public String getSubref() {
        return subref;
    }

    public String getHost() {
        return host;
    }

    public String getServer() {
        return server;
    }

    @Override
    public String toString() {
        return "Task{" +
                "address=" + address +
                ", attribute='" + attribute + '\'' +
                '}';
    }
} public class BufferedStorageDispatcher implements Scheduler.CompletionHandler {

    private static final int MAX_BATCH_SIZE = 24;
    private static final int BUFFER_SIZE = 100;
    private final StorageAdapter storageAdapter;
    private final Diagnostics diagnostics;
    private final BlockingQueue<DataPoint> queue;
    private final Worker worker;

    public BufferedStorageDispatcher(StorageAdapter storageAdapter, Diagnostics diagnostics) {
        this.storageAdapter = storageAdapter;
        this.diagnostics = diagnostics;
        this.queue = new ArrayBlockingQueue<DataPoint>(BUFFER_SIZE);
        this.worker = new Worker(queue);
    }

    public void start() {
        worker.start();
    }

    public void shutdown() {
        worker.setKeepRunnig(false);
    }

    @Override
    public void onCompleted(DataPoint sample) {
        if(queue.remainingCapacity()>0) {
            //System.out.println(sample.getTask().getAttribute()+" > "+sample.getValue());
            diagnostics.getStorageBufferSize().inc();
            queue.add(sample);
        }
        else {
            throw new RuntimeException("buffer capacity exceeded");
        }
    }

    @Override
    public void onFailed(Throwable e) {
        e.printStackTrace();
    }

    public class Worker extends Thread {
        private final BlockingQueue<DataPoint> queue;
        private boolean keepRunning = true;

        public Worker(BlockingQueue<DataPoint> queue) {
            this.queue = queue;
        }

        public void run() {
            try {
                while ( keepRunning ) {

                    // batch processing
                    DataPoint sample = queue.take();
                    Set<DataPoint> samples = new HashSet<>();
                    queue.drainTo(samples, MAX_BATCH_SIZE);
                    samples.add(sample);

                    diagnostics.getStorageBufferSize().dec(samples.size());

                    // dispatch
                    storageAdapter.store(samples);
                }
            }
            catch ( InterruptedException ie ) {
                // just terminate
            }
        }

        public void setKeepRunnig(boolean keepRunning) {
            this.keepRunning = keepRunning;
        }
    }
} public class InfluxStorageAdapter implements StorageAdapter {

    private InfluxDB influxDB;
    private String dbName;
    private Diagnostics diagnostics;
    private Configuration config;
    private DefaultKeyResolution keyResolution;

    @Override
    public void setConfiguration(Configuration config) {
        this.config = config;
        this.influxDB = InfluxDBFactory.connect(
                config.getStorageUrl(),
                config.getStorageUser(),
                config.getStoragePassword()
        );
        this.dbName = config.getStorageDBName();

        this.keyResolution = new DefaultKeyResolution();
    }

    @Override
    public void setDiagnostics(Diagnostics diag) {
        this.diagnostics = diag;
    }

    @Override
    public void store(Set<DataPoint> datapoints) {

        try {

            Serie[] series = new Serie[datapoints.size()];
            int i=0;
            for (DataPoint datapoint : datapoints) {

                Task task = datapoint.getTask();
                String key = keyResolution.resolve(task);
                Serie dataPoint = new Serie.Builder(key)
                        .columns("datapoint")
                        .values(datapoint.getValue())
                        .build();

                series[i] = dataPoint;
                i++;
            }

            this.influxDB.write(this.dbName, TimeUnit.MILLISECONDS, series);

        } catch (Throwable t) {
            diagnostics.getStorageErrorRate().mark(1);
            t.printStackTrace();
        }

    }
} public class RHQStorageAdapter implements StorageAdapter {

    private Configuration config;
    private Diagnostics diagnostics;
    private final HttpClient httpclient;
    private final DefaultKeyResolution keyResolution;

    public RHQStorageAdapter() {
        this.httpclient = new DefaultHttpClient();
        this.keyResolution = new DefaultKeyResolution();
    }

    @Override
    public void setConfiguration(Configuration config) {
        this.config = config;
    }

    @Override
    public void setDiagnostics(Diagnostics diag) {
        this.diagnostics = diag;
    }

    @Override
    public void store(Set<DataPoint> datapoints) {
        HttpPost post = new HttpPost(config.getStorageUrl());
        try {
            List<SingleMetric> metrics = new ArrayList<>();

            for (DataPoint datapoint : datapoints) {
                Task task = datapoint.getTask();
                String key = keyResolution.resolve(task);
                metrics.add(new SingleMetric(key, datapoint.getTimestamp(), datapoint.getValue()));
            }


            // If we have data, send it to the RHQ Metrics server

            if (metrics.size()>0) {
                post.setHeader("Content-Type", "application/json;charset=utf-8");
                post.setEntity(new StringEntity(Batcher.metricListToJson(metrics)));

                HttpResponse httpResponse = httpclient.execute(post);
                StatusLine statusLine = httpResponse.getStatusLine();

                if (statusLine.getStatusCode() != 200) {
                    throw new Exception("HTTP Status "+statusLine.getStatusCode()+": "+statusLine);
                }


            }
        } catch (Throwable t) {
            t.printStackTrace();
            diagnostics.getStorageErrorRate().mark(1);
        }
        finally {
            post.releaseConnection();
        }

    }
} public interface StorageAdapter {
    void store(Set<DataPoint> datapoints);
    void setConfiguration(Configuration config);
    void setDiagnostics(Diagnostics diag);
}
import com.codahale.metrics.ConsoleReporter;
import com.codahale.metrics.Counter;
import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.ScheduledReporter;
import com.codahale.metrics.Timer;
import org.wildfly.metrics.scheduler.config.Address;
import org.wildfly.metrics.scheduler.config.Configuration;
import org.wildfly.metrics.scheduler.config.ResourceRef;
import org.wildfly.metrics.scheduler.diagnose.Diagnostics;
import org.wildfly.metrics.scheduler.diagnose.StorageReporter;
import org.wildfly.metrics.scheduler.polling.ClientFactoryImpl;
import org.wildfly.metrics.scheduler.polling.IntervalBasedScheduler;
import org.wildfly.metrics.scheduler.polling.Scheduler;
import org.wildfly.metrics.scheduler.polling.Task;
import org.wildfly.metrics.scheduler.storage.BufferedStorageDispatcher;
import org.wildfly.metrics.scheduler.storage.InfluxStorageAdapter;
import org.wildfly.metrics.scheduler.storage.RHQStorageAdapter;
import org.wildfly.metrics.scheduler.storage.StorageAdapter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import static com.codahale.metrics.MetricRegistry.name;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
package org.wildfly.metrics.scheduler;




/**
 * The core service that creates task lists from a {@link org.wildfly.metrics.scheduler.config.Configuration}
 * and schedules work through a {@link org.wildfly.metrics.scheduler.polling.Scheduler}.
 * The resulting data will be pushed to a {@link org.wildfly.metrics.scheduler.storage.StorageAdapter}
 *
 * @author Heiko Braun
 * @since 10/10/14
 */
public class Service implements TopologyChangeListener {

    private final StorageAdapter storageAdapter;
    private Configuration configuration;
    private Scheduler scheduler;
    private Diagnostics diagnostics;
    private ScheduledReporter reporter;
    private boolean started = false;
    private BufferedStorageDispatcher completionHandler;

    /**
        *
        * @param configuration
        */
    public Service(Configuration configuration) {
        this(configuration, new ClientFactoryImpl(
                configuration.getHost(),
                configuration.getPort(),
                configuration.getUsername(),
                configuration.getPassword())
        );
    }

    /**
     *
     * @param configuration
     * @param clientFactory
     */
    public Service(Configuration configuration, ModelControllerClientFactory clientFactory) {

        this.configuration = configuration;

        final MetricRegistry metrics = new MetricRegistry();

        this.diagnostics = createDiagnostics(metrics);

        if(Configuration.Storage.RHQ == configuration.getStorageAdapter())
        {
            this.storageAdapter = new RHQStorageAdapter();