package org.rhq.metrics;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.google.common.util.concurrent.ListenableFuture;

import org.rhq.metrics.core.MetricsService;
import org.rhq.metrics.core.RawNumericMetric;
import org.rhq.metrics.impl.cassandra.MetricsServiceCassandra;
import org.rhq.metrics.impl.memory.MemoryMetricsService;

/**
 * @author John Sanda
 */
public class RHQMetrics {

    public static class Builder {

        private boolean usingCassandra;

        private Map<String, String> options;

        public Builder() {
            String cqlPortString = System.getenv("CQL_PORT");
            if(cqlPortString == null){
                cqlPortString = "9042";
            }

            options = new HashMap<>();
            options.put("cqlport", cqlPortString);
            options.put("nodes", "127.0.0.1");
            options.put("keyspace", "rhq-metrics");
        }

        public Builder withOptions(Map<String,String> options) {
            this.options.putAll(options);
            return this;
        }

        public Builder withInMemoryDataStore() {
            usingCassandra = false;
            return this;
        }

        public Builder withCassandraDataStore() {
            usingCassandra = true;
            return this;
        }

        public Builder withCQLPort(int port) {
            options.put("cqlport", Integer.toString(port));
            return this;
        }

        public Builder withKeyspace(String keyspace) {
            options.put("keyspace", keyspace);
            return this;
        }

        public Builder withNodes(String... nodes) {
            StringBuilder buffer = new StringBuilder();
            for (String node : nodes) {
                buffer.append(node).append(",");
            }
            if (buffer.length() > 0) {
                buffer.deleteCharAt(buffer.length() - 1);
            }
            options.put("nodes", buffer.toString());
            return this;
        }

        public RHQMetrics build() {
            MetricsService metricsService;

            if (usingCassandra) {
                metricsService = new MetricsServiceCassandra();
            } else {
                metricsService = new MemoryMetricsService();
            }
            metricsService.startUp(options);

            return new RHQMetrics(metricsService);
        }

    }

    private MetricsService metricsService;

    private RHQMetrics(MetricsService metricsService) {
        this.metricsService = metricsService;
    }

    public ListenableFuture<Void> addData(RawNumericMetric data) {
        return metricsService.addData(data);
    }

    public ListenableFuture<Map<RawNumericMetric, Throwable>> addData(Set<RawNumericMetric> data) {
        return metricsService.addData(data);
    }

    public ListenableFuture<List<RawNumericMetric>> findData(String id, long start, long end) {
        return metricsService.findData(id, start, end);
    }

    public void shutdown() {
        metricsService.shutdown();
    }

}

@@ -610,16 +610,25 @@
 String c
+assandraC
 qlPortSt
@@ -649,16 +649,26 @@
 getenv(%22
+CASSANDRA_
 CQL_PORT
@@ -689,10 +689,20 @@
   if
+ 
 (c
+assandraC
 qlPo
@@ -718,16 +718,17 @@
 == null)
+ 
 %7B%0A      
@@ -738,16 +738,25 @@
        c
+assandraC
 qlPortSt
@@ -777,32 +777,205 @@
 %0A            %7D%0A%0A
+            String cassandraNodes = System.getenv(%22CASSANDRA_NODES%22);%0A            if (cassandraNodes == null) %7B%0A                cassandraNodes = %22127.0.0.1%22;%0A            %7D%0A%0A
             opti
@@ -1033,16 +1033,25 @@
 port%22, c
+assandraC
 qlPortSt
@@ -1090,27 +1090,30 @@
 nodes%22, 
-%22127.0.0.1%22
+cassandraNodes
 );%0A     

Update metrics storage configurator to use environment variables for CQL port and node addresses.