Index: core/metrics-core-impl/src/main/java/org/hawkular/metrics/core/impl/GaugeDataPointCollector.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- core/metrics-core-impl/src/main/java/org/hawkular/metrics/core/impl/GaugeDataPointCollector.java	(revision 4bb1f5e26510c755e06287fa1e980c7c137ac176)
+++ core/metrics-core-impl/src/main/java/org/hawkular/metrics/core/impl/GaugeDataPointCollector.java	(revision )
@@ -16,6 +16,8 @@
  */
 package org.hawkular.metrics.core.impl;
 
+import java.util.List;
+
 import org.apache.commons.math3.stat.descriptive.moment.Mean;
 import org.apache.commons.math3.stat.descriptive.rank.Max;
 import org.apache.commons.math3.stat.descriptive.rank.Min;
@@ -38,13 +40,19 @@
     private PSquarePercentile median = new PSquarePercentile(50.0);
     private Max max = new Max();
     private PSquarePercentile percentile95th = new PSquarePercentile(95.0);
+    private int index;
 
     GaugeDataPointCollector(Buckets buckets, int bucketIndex) {
         this.buckets = buckets;
         this.bucketIndex = bucketIndex;
     }
 
+    public GaugeDataPointCollector(Buckets buckets) {
+        this(buckets, -1);
+    }
+
     void increment(DataPoint<Double> dataPoint) {
+        index = buckets.getIndex(dataPoint.getTimestamp());
         Double value = dataPoint.getValue();
         min.increment(value);
         average.increment(value);
@@ -53,8 +61,12 @@
         percentile95th.increment(value);
     }
 
+    void increment(List<DataPoint<Double>> dataPoints) {
+        dataPoints.forEach(this::increment);
+    }
+
     GaugeBucketPoint toBucketPoint() {
-        long from = buckets.getBucketStart(bucketIndex);
+        long from = buckets.getBucketStart(bucketIndex != -1 ? bucketIndex : index);
         long to = from + buckets.getStep();
         return new GaugeBucketPoint.Builder(from, to)
                 .setMin(min.getResult())
Index: rest-tests/src/test/groovy/org/hawkular/metrics/rest/GaugeMetricStatisticsITest.groovy
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- rest-tests/src/test/groovy/org/hawkular/metrics/rest/GaugeMetricStatisticsITest.groovy	(revision 4bb1f5e26510c755e06287fa1e980c7c137ac176)
+++ rest-tests/src/test/groovy/org/hawkular/metrics/rest/GaugeMetricStatisticsITest.groovy	(revision )
@@ -17,9 +17,12 @@
 package org.hawkular.metrics.rest
 
 import static java.lang.Double.NaN
+import static java.util.concurrent.TimeUnit.DAYS
+import static java.util.concurrent.TimeUnit.MILLISECONDS
 
 import static org.joda.time.Seconds.seconds
 import static org.junit.Assert.assertEquals
+import static org.junit.Assert.assertTrue
 
 import org.apache.commons.math3.stat.descriptive.moment.Mean
 import org.apache.commons.math3.stat.descriptive.rank.Max
@@ -30,6 +33,8 @@
 import org.joda.time.Duration
 import org.junit.Test
 
+import com.google.common.base.Stopwatch
+
 /**
  * @author Thomas Segismont
  */
@@ -159,12 +164,14 @@
   void largeDataSetTest() {
     String tenantId = nextTenantId()
     String metric = "test"
-    int nbOfBuckets = 10
-    long bucketSize = Duration.standardDays(1).millis
-    int interval = Duration.standardMinutes(1).millis
+    int nbOfBuckets = 1
+    long bucketSize = Duration.standardHours(1).millis
+    int interval = Duration.standardSeconds(1).millis
     int sampleSize = (bucketSize / interval) - 1
 
-    def start = new DateTimeService().currentHour().minus(3 * nbOfBuckets * bucketSize)
+    def hour = new DateTimeService().currentHour()
+    def back = 3 * nbOfBuckets * bucketSize
+    def start = hour.minus(back)
 
     def expectedData = []
 
@@ -193,6 +200,8 @@
           body: data, headers: [(tenantHeaderName): tenantId])
       assertEquals(200, response.status)
 
+      if (step % 10 == 0) println "Finished posting for step ${step}"
+
       expectedData.add([
           start         : bucketStart.millis,
           end           : bucketStart.plus(bucketSize).millis,
@@ -205,11 +214,17 @@
       ])
     }
 
+    println "Finished posting"
+
+    def stopwatch = Stopwatch.createStarted()
     def response = hawkularMetrics.get(path: "gauges/$metric/data",
         query: [
             start: start.millis, end: start.plus(nbOfBuckets * bucketSize).millis, bucketDuration: "${bucketSize}ms"
         ],
         headers: [(tenantHeaderName): tenantId])
+    stopwatch.stop()
+    println "time: ${stopwatch}"
+
     assertEquals(200, response.status)
 
     assertEquals('The number of bucketed data points is wrong', expectedData.size(), response.data.size())
Index: api/metrics-api-jaxrs/src/main/java/org/hawkular/metrics/api/jaxrs/handler/GaugeHandler.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- api/metrics-api-jaxrs/src/main/java/org/hawkular/metrics/api/jaxrs/handler/GaugeHandler.java	(revision 4bb1f5e26510c755e06287fa1e980c7c137ac176)
+++ api/metrics-api-jaxrs/src/main/java/org/hawkular/metrics/api/jaxrs/handler/GaugeHandler.java	(revision )
@@ -63,7 +63,10 @@
 import org.hawkular.metrics.core.api.Metric;
 import org.hawkular.metrics.core.api.MetricId;
 import org.hawkular.metrics.core.api.MetricsService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import com.google.common.base.Stopwatch;
 import com.wordnik.swagger.annotations.Api;
 import com.wordnik.swagger.annotations.ApiOperation;
 import com.wordnik.swagger.annotations.ApiParam;
@@ -81,6 +84,8 @@
 @Produces(APPLICATION_JSON)
 @Api(value = "", description = "Gauge metrics interface")
 public class GaugeHandler {
+    private static final Logger logger = LoggerFactory.getLogger(GaugeHandler.class);
+
     private static final long EIGHT_HOURS = MILLISECONDS.convert(8, HOURS);
 
     @Inject
@@ -290,8 +295,10 @@
                 return;
             }
 
+            Stopwatch stopwatch = Stopwatch.createStarted();
             metricsService.findGaugeStats(metricId, startTime, endTime, buckets)
                     .map(ApiUtils::collectionToResponse)
+                    .doOnCompleted(() -> logger.error("grand total = " + stopwatch))
                     .subscribe(asyncResponse::resume, t -> asyncResponse.resume(ApiUtils.serverError(t)));
         }
     }
Index: api/metrics-api-common/src/main/java/org/hawkular/metrics/api/jaxrs/util/ApiUtils.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- api/metrics-api-common/src/main/java/org/hawkular/metrics/api/jaxrs/util/ApiUtils.java	(revision 4bb1f5e26510c755e06287fa1e980c7c137ac176)
+++ api/metrics-api-common/src/main/java/org/hawkular/metrics/api/jaxrs/util/ApiUtils.java	(revision )
@@ -43,6 +43,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.base.Stopwatch;
 import com.google.common.base.Throwables;
 
 import rx.Observable;
@@ -57,7 +58,11 @@
     }
 
     public static Response collectionToResponse(Collection<?> collection) {
-        return collection.isEmpty() ? noContent() : Response.ok(collection).type(MediaType.APPLICATION_JSON).build();
+        Stopwatch stopwatch = Stopwatch.createStarted();
+        Response response = collection.isEmpty() ? noContent() : Response.ok(collection).type(MediaType
+                .APPLICATION_JSON).build();
+        LOG.error("reponse creation time = " + stopwatch);
+        return response;
     }
 
     public static Response serverError(Throwable t, String message) {
\ No newline at end of file
Index: core/metrics-core-impl/src/main/java/org/hawkular/metrics/core/impl/MetricsServiceImpl.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- core/metrics-core-impl/src/main/java/org/hawkular/metrics/core/impl/MetricsServiceImpl.java	(revision 4bb1f5e26510c755e06287fa1e980c7c137ac176)
+++ core/metrics-core-impl/src/main/java/org/hawkular/metrics/core/impl/MetricsServiceImpl.java	(revision )
@@ -18,7 +18,9 @@
 
 import static java.util.Collections.emptyMap;
 import static java.util.Comparator.comparingLong;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.MINUTES;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
 
 import static org.hawkular.metrics.core.api.MetricType.AVAILABILITY;
 import static org.hawkular.metrics.core.api.MetricType.COUNTER;
@@ -45,6 +47,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
 import java.util.function.Predicate;
 import java.util.regex.Pattern;
 
@@ -82,6 +85,7 @@
 import com.datastax.driver.core.ResultSetFuture;
 import com.datastax.driver.core.Row;
 import com.datastax.driver.core.Session;
+import com.google.common.base.Stopwatch;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.util.concurrent.FutureCallback;
@@ -94,6 +98,7 @@
 import rx.functions.Func1;
 import rx.functions.Func2;
 import rx.functions.Func3;
+import rx.subjects.PublishSubject;
 
 /**
  * @author John Sanda
@@ -675,9 +680,25 @@
         Timer timer = getDataPointFindTimer(metricType);
         Func3<MetricId<T>, Long, Long, Observable<ResultSet>> finder = getDataPointFinder(metricType);
         Func1<Row, DataPoint<T>> mapper = getDataPointMapper(metricType);
+        LongAdder mappingTime = new LongAdder();
+        Func1<Row, DataPoint<T>> mapper1 = row -> {
+            Stopwatch mappingTimeLocal = Stopwatch.createStarted();
+            DataPoint<T> point = mapper.call(row);
+            mappingTimeLocal.stop();
+            mappingTime.add(mappingTimeLocal.elapsed(NANOSECONDS));
+            return point;
+        };
         return time(timer, () -> finder.call(metricId, start, end)
-                .flatMap(Observable::from)
-                .map(mapper));
+                .flatMap(rows -> {
+//                    Stopwatch stopwatch = Stopwatch.createStarted();
+//                    Observable<Row> observable = Observable.from(rows.all());
+//                    logger.error("loading = " + stopwatch);
+                    Observable<Row> observable = Observable.from(rows);
+                    return observable;
+                })
+                .map(mapper1)).doOnCompleted(() -> {
+            logger.error("mapping time = " + MILLISECONDS.convert(mappingTime.sum(), NANOSECONDS) + " ms");
+        });
     }
 
     private <T> Timer getDataPointFindTimer(MetricType<T> metricType) {
@@ -803,13 +824,41 @@
     @Override
     public Observable<List<GaugeBucketPoint>> findGaugeStats(MetricId<Double> metricId, long start, long end,
                                                              Buckets buckets) {
+//        Stopwatch toList = Stopwatch.createStarted();
+//        List<DataPoint<Double>> dataPoints = findDataPoints(metricId, start, end).toList().toBlocking().single();
+//        logger.error("toList = " + toList);
+        PublishSubject<DataPoint<Double>> subject = PublishSubject.create();
+//        Stopwatch rx = Stopwatch.createStarted();
         return findDataPoints(metricId, start, end)
-                .groupBy(dataPoint -> buckets.getIndex(dataPoint.getTimestamp()))
-                .flatMap(group -> group.collect(() -> new GaugeDataPointCollector(buckets, group.getKey()),
-                        GaugeDataPointCollector::increment))
+                .doOnNext(v -> {
+                    //logger.error("doOnNext v = " + v);
+                    subject.onNext(v);
+                })
+                .window(subject.map(DataPoint::getTimestamp).map(t -> buckets.getIndex(t)).distinctUntilChanged()
+//                        .doOnNext(v -> {
+//                            logger.error("boundary onNext v = " + v);
+//                        })
+                )
+                .flatMap(window -> {
+                    return window.collect(() -> new GaugeDataPointCollector(buckets),
+                            GaugeDataPointCollector::increment);
+//                    return window.buffer(100).collect(() -> new GaugeDataPointCollector(buckets),
+//                            GaugeDataPointCollector::increment);
+                })
                 .map(GaugeDataPointCollector::toBucketPoint)
                 .toMap(GaugeBucketPoint::getStart)
-                .map(pointMap -> GaugeBucketPoint.toList(pointMap, buckets));
+                .map(pointMap -> GaugeBucketPoint.toList(pointMap, buckets))/*.doOnCompleted(() -> {
+                    logger.error("rx = " + rx);
+                })*/;
+//        return Observable.from(dataPoints)
+//                .groupBy(dataPoint -> buckets.getIndex(dataPoint.getTimestamp()))
+//                .flatMap(group -> group.collect(() -> new GaugeDataPointCollector(buckets, group.getKey()),
+//                        GaugeDataPointCollector::increment))
+//                .map(GaugeDataPointCollector::toBucketPoint)
+//                .toMap(GaugeBucketPoint::getStart)
+//                .map(pointMap -> GaugeBucketPoint.toList(pointMap, buckets)).doOnCompleted(() -> {
+//                    logger.error("rx = " + rx);
+//                });
     }
 
     @Override
