// Path: src/main/java/net/iponweb/disthene/bean/Metric.java
// public class Metric {
// 
//     private MetricKey key;
//     private double value;
// 
//     public Metric(String input, Rollup rollup) {
//         String[] splitInput = input.split("\\s");
//         // We were interning tenant and path here - we are going to store them all (or almost so) constantly anyhow in multiple places
//         // In fact this also work for a moderate metrics stream. Once we start receiving 10s of millions different metrics, it tends to degrade quite a bit
//         // So, leaving this only for tenant
//         this.key = new MetricKey(
//                 splitInput.length >=4 ? splitInput[3].intern() : "NONE",
//                 splitInput[0],
//                 rollup.getRollup(),
//                 rollup.getPeriod(),
//                 normalizeTimestamp(Long.parseLong(splitInput[2]), rollup));
//         this.value = Double.parseDouble(splitInput[1]);
//     }
// 
//     public Metric(String tenant, String path, int rollup, int period, double value, long timestamp) {
//         this.key = new MetricKey(tenant, path, rollup, period, timestamp);
//         this.value = value;
//     }
// 
//     public Metric(MetricKey key, double value) {
//         this.key = key;
//         this.value = value;
//     }
// 
//     private long normalizeTimestamp(long timestamp, Rollup rollup) {
//         return (timestamp / rollup.getRollup()) * rollup.getRollup();
//     }
// 
//     public String getId() {
//         return getTenant() + "_" + getPath();
//     }
// 
//     public String getTenant() {
//         return key.getTenant();
//     }
// 
//     public String getPath() {
//         return key.getPath();
//     }
// 
//     public int getRollup() {
//         return key.getRollup();
//     }
// 
//     public int getPeriod() {
//         return key.getPeriod();
//     }
// 
//     public double getValue() {
//         return value;
//     }
// 
//     public void setValue(double value) {
//         this.value = value;
//     }
// 
//     public long getTimestamp() {
//         return key.getTimestamp();
//     }
// 
//     @Override
//     public String toString() {
//         return "Metric{" +
//                 "key=" + key +
//                 ", value=" + value +
//                 '}';
//     }
// }
// 
// Path: src/main/java/net/iponweb/disthene/events/DistheneEvent.java
// public interface DistheneEvent {
// }
// 
// Path: src/main/java/net/iponweb/disthene/events/StoreErrorEvent.java
// public class StoreErrorEvent implements DistheneEvent {
// 
//     private int count;
// 
//     public StoreErrorEvent(int count) {
//         this.count = count;
//     }
// 
//     public int getCount() {
//         return count;
//     }
// }
// 
// Path: src/main/java/net/iponweb/disthene/events/StoreSuccessEvent.java
// public class StoreSuccessEvent implements DistheneEvent{
// 
//     private int count;
// 
//     public StoreSuccessEvent(int count) {
//         this.count = count;
//     }
// 
//     public int getCount() {
//         return count;
//     }
// }


import com.datastax.driver.core.*;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import net.engio.mbassy.bus.MBassador;
import net.iponweb.disthene.bean.Metric;
import net.iponweb.disthene.events.DistheneEvent;
import net.iponweb.disthene.events.StoreErrorEvent;
import net.iponweb.disthene.events.StoreSuccessEvent;
import org.apache.log4j.Logger;
import java.util.*;
import java.util.concurrent.Executor;
                            metric.getTenant(),
                            metric.getRollup(),
                            metric.getPeriod(),
                            metric.getPath(),
                            metric.getTimestamp()
                )
        );

        if (statements.size() >= batchSize || (lastFlushTimestamp < System.currentTimeMillis() - INTERVAL)) {
            lastFlushTimestamp = System.currentTimeMillis();
            flush();
        }
    }

    private void flush() {
        List<List<Statement>> batches = splitByToken();

        for (List<Statement> batchStatements : batches) {
            BatchStatement batch = new BatchStatement(BatchStatement.Type.UNLOGGED);
            final int batchSize = batchStatements.size();

            for (Statement s : batchStatements) {
                batch.add(s);
            }

            ResultSetFuture future = session.executeAsync(batch);
            Futures.addCallback(future,
                    new FutureCallback<ResultSet>() {
                        @Override
                        public void onSuccess(ResultSet result) {
// Path: src/main/java/net/iponweb/disthene/bean/Metric.java
// public class Metric {
// 
//     private MetricKey key;
//     private double value;
// 
//     public Metric(String input, Rollup rollup) {
//         String[] splitInput = input.split("\\s");
//         // We were interning tenant and path here - we are going to store them all (or almost so) constantly anyhow in multiple places
//         // In fact this also work for a moderate metrics stream. Once we start receiving 10s of millions different metrics, it tends to degrade quite a bit
//         // So, leaving this only for tenant
//         this.key = new MetricKey(
//                 splitInput.length >=4 ? splitInput[3].intern() : "NONE",
//                 splitInput[0],
//                 rollup.getRollup(),
//                 rollup.getPeriod(),
//                 normalizeTimestamp(Long.parseLong(splitInput[2]), rollup));
//         this.value = Double.parseDouble(splitInput[1]);
//     }
// 
//     public Metric(String tenant, String path, int rollup, int period, double value, long timestamp) {
//         this.key = new MetricKey(tenant, path, rollup, period, timestamp);
//         this.value = value;
//     }
// 
//     public Metric(MetricKey key, double value) {
//         this.key = key;
//         this.value = value;
//     }
// 
//     private long normalizeTimestamp(long timestamp, Rollup rollup) {
//         return (timestamp / rollup.getRollup()) * rollup.getRollup();
//     }
// 
//     public String getId() {
//         return getTenant() + "_" + getPath();
//     }
// 
//     public String getTenant() {
//         return key.getTenant();
//     }
// 
//     public String getPath() {
//         return key.getPath();
//     }
// 
//     public int getRollup() {
//         return key.getRollup();
//     }
// 
//     public int getPeriod() {
//         return key.getPeriod();
//     }
// 
//     public double getValue() {
//         return value;
//     }
// 
//     public void setValue(double value) {
//         this.value = value;
//     }
// 
//     public long getTimestamp() {
//         return key.getTimestamp();
//     }
// 
//     @Override
//     public String toString() {
//         return "Metric{" +
//                 "key=" + key +
//                 ", value=" + value +
//                 '}';
//     }
// }
// 
// Path: src/main/java/net/iponweb/disthene/events/DistheneEvent.java
// public interface DistheneEvent {
// }
// 
// Path: src/main/java/net/iponweb/disthene/events/StoreErrorEvent.java
// public class StoreErrorEvent implements DistheneEvent {
// 
//     private int count;
// 
//     public StoreErrorEvent(int count) {
//         this.count = count;
//     }
// 
//     public int getCount() {
//         return count;
//     }
// }
// 
// Path: src/main/java/net/iponweb/disthene/events/StoreSuccessEvent.java
// public class StoreSuccessEvent implements DistheneEvent{
// 
//     private int count;
// 
//     public StoreSuccessEvent(int count) {
//         this.count = count;
//     }
// 
//     public int getCount() {
//         return count;
//     }
// }


// Path: src/main/java/net/iponweb/disthene/service/store/BatchWriterThread.java
import com.datastax.driver.core.*;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import net.engio.mbassy.bus.MBassador;
import net.iponweb.disthene.bean.Metric;
import net.iponweb.disthene.events.DistheneEvent;
import net.iponweb.disthene.events.StoreErrorEvent;
import net.iponweb.disthene.events.StoreSuccessEvent;
import org.apache.log4j.Logger;
import java.util.*;
import java.util.concurrent.Executor;

                            metric.getTenant(),
                            metric.getRollup(),
                            metric.getPeriod(),
                            metric.getPath(),
                            metric.getTimestamp()
                )
        );

        if (statements.size() >= batchSize || (lastFlushTimestamp < System.currentTimeMillis() - INTERVAL)) {
            lastFlushTimestamp = System.currentTimeMillis();
            flush();
        }
    }

    private void flush() {
        List<List<Statement>> batches = splitByToken();

        for (List<Statement> batchStatements : batches) {
            BatchStatement batch = new BatchStatement(BatchStatement.Type.UNLOGGED);
            final int batchSize = batchStatements.size();

            for (Statement s : batchStatements) {
                batch.add(s);
            }

            ResultSetFuture future = session.executeAsync(batch);
            Futures.addCallback(future,
                    new FutureCallback<ResultSet>() {
                        @Override
                        public void onSuccess(ResultSet result) {
                            bus.post(new StoreSuccessEvent(batchSize)).now();