flatmaptest.java

@Test
    public void addAndFetchGaugeDataAggregates() throws Exception {
        DateTime start = now().minusMinutes(30);
        DateTime end = start.plusMinutes(20);
        String tenantId = "t1";

        metricsService.createTenant(new Tenant(tenantId)).toBlocking().lastOrDefault(null);

        Metric<Double> m1 = new Metric<>(new MetricId<>(tenantId, GAUGE, "m1"), asList(
                new DataPoint<>(start.getMillis(), 10.0),
                new DataPoint<>(start.plusMinutes(2).getMillis(), 20.0),
                new DataPoint<>(start.plusMinutes(4).getMillis(), 30.0),
                new DataPoint<>(end.getMillis(), 40.0)
        ));

        Observable<Void> insertObservable = metricsService.addDataPoints(GAUGE, Observable.just(m1));
        insertObservable.toBlocking().lastOrDefault(null);

        Func1<Observable<DataPoint<Double>>, Observable<Double>> min = data -> {
            logger.debug("Executing MIN");
            return Aggregate.Min.call(data);
        };
        Func1<Observable<DataPoint<Double>>, Observable<Double>> max = data -> {
            logger.debug("Executing MAX");
            return Aggregate.Max.call(data);
        };
        Func1<Observable<DataPoint<Double>>, Observable<Double>> avg = data -> {
            logger.debug("Executing AVG");
            return Aggregate.Average.call(data);
        };
        Func1<Observable<DataPoint<Double>>, Observable<Double>> sum = data -> {
            logger.debug("Executing SUM");
            return Aggregate.Sum.call(data);
        };

        Observable<Double> observable = metricsService
                .findGaugeData(new MetricId<>(tenantId, GAUGE, "m1"), start.getMillis(), (end.getMillis() + 1000),
                        min, max, avg, sum);
        List<Double> actual = toList(observable);
        List<Double> expected = asList(10.0, 40.0, 25.0, 100.0);

        assertEquals(actual, expected, "The data does not match the expected values");
    }

    private <T> List<T> toList(Observable<T> observable) {
        return ImmutableList.copyOf(observable.toBlocking().toIterable());
    }