aggregate.java

public <T> T applyFunction(String tenantId, MetricId id, long start, long end,
            Func1<Observable<DataPoint<Double>>, T> function) {
        Observable<DataPoint<Double>> dataPoints = metricsService.findGaugeData(tenantId, id, start, end);
        return function.call(dataPoints);
    }

class Aggregate {
        public final Double min;
        public final Double max;
        public final Double avg;

        public Aggregate(Double min, Double max, Double avg) {
            this.min = min;
            this.max = max;
            this.avg = avg;
        }
    }

    @Test
    public void getAggregate() {
        DateTime start = now().minusMinutes(30);
        DateTime end = start.plusMinutes(20);
        String tenantId = "t1";
        MetricId id = new MetricId("m1");

        Func1<Observable<DataPoint<Double>>, Observable<Aggregate>> compute = dataPoints -> {
            Observable<Double> values = dataPoints.map(DataPoint::getValue).cache();
            Observable<Double> avg = MathObservable.averageDouble(values);
            Observable<Double> max = MathObservable.max(values);
            Observable<Double> min = MathObservable.min(values);

            return Observable.zip(min, max, avg, Aggregate::new);
        };

        Observable<Aggregate> aggregate = applyFunction(tenantId, id, start.getMillis(), end.getMillis(), compute);
    }

The compute function maps the data points to get just the values and caches them. It then calculates the max, min, and average and combines them using the zip operator. There is some RxJava stuff going on here, but the key thing is higher order functions. Rather than have a bunch of methods that operate on the stream of data points, I have a single method that applies a function over those data points. The method does not care what the function does or returns, and the function does not care how it receives the data points. This separation of concerns make things more loosely coupled and makes it easier to test as well.