/*
 * Copyright 2014-2015 Red Hat, Inc. and/or its affiliates
 * and other contributors as indicated by the @author tags.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.hawkular.metrics.core.impl;

import static java.util.Collections.singletonList;

import static org.hawkular.metrics.core.api.MetricType.COUNTER;
import static org.hawkular.metrics.core.api.MetricType.COUNTER_RATE;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.hawkular.metrics.core.api.DataPoint;
import org.hawkular.metrics.core.api.Metric;
import org.hawkular.metrics.core.api.MetricId;
import org.hawkular.metrics.core.api.MetricsService;
import org.hawkular.metrics.tasks.api.Task2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import rx.Observable;
import rx.functions.Action1;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import static java.util.Collections.singletonList;
import static org.hawkular.metrics.core.api.MetricType.COUNTER;
import static org.hawkular.metrics.core.api.MetricType.COUNTER_RATE;

/**
 * Calculates and persists rates for all counter metrics of a single tenant.
 */
public class GenerateRate implements Action1<Task2> {

    private static final Logger logger = LoggerFactory.getLogger(GenerateRate.class);

    private MetricsService metricsService;

    public GenerateRate(MetricsService metricsService) {
        this.metricsService = metricsService;
    }

    @Override
    public void call(Task2 task) {
        // TODO We need to make this fault tolerant. See HWKMETRICS-213 for details.
        logger.info("Generating rate for {}", task);
        String tenant = task.getParameters().get("tenant");
        long start = task.getTrigger().getTriggerTime();
        long end = start + TimeUnit.MINUTES.toMillis(1);

        Observable<Metric<Double>> rates = metricsService.findMetrics(tenant, COUNTER)
                .flatMap(counter -> metricsService.findCounterData(counter.getId(), start, end)
                        .take(1)
                        .map(dataPoint -> ((dataPoint.getValue().doubleValue() / (end - start) * 1000)))
                        .map(rate -> new Metric<>(new MetricId(tenant, COUNTER_RATE, counter.getId().getName()),
                                singletonList(new DataPoint<>(start, rate)))));
        Observable<Void> updates = metricsService.addGaugeData(rates);

        CountDownLatch latch = new CountDownLatch(1);

        logger.debug("start = {}, end = {}", start, end);
        metricsService.findCounterData(id, start, end)
                .take(1)
                .doOnNext(dataPoint -> logger.debug("Data Point = {}", dataPoint))
                .map(dataPoint -> ((dataPoint.getValue().doubleValue() / (end - start) * 1000)))
                .map(rate -> new Metric<>(new MetricId(id.getTenantId(), COUNTER_RATE, id.getName()),
                        singletonList(new DataPoint<>(start, rate))))
                .flatMap(metric -> metricsService.addGaugeData(Observable.just(metric)))
                .subscribe(
                        aVoid -> {
                        },
                        t -> {
                            logger.warn("Failed to persist rate data", t);
                            latch.countDown();
                        },
                        () -> {
                            logger.debug("Successfully persisted rate data");
                            latch.countDown();
                        }
                );
        try {
            latch.await();
        } catch (InterruptedException e) {
        }
    }

}

@@ -1338,276 +1338,8 @@
 1;%0A%0A
-import java.util.concurrent.CountDownLatch;%0Aimport java.util.concurrent.TimeUnit;%0A%0Aimport static java.util.Collections.singletonList;%0Aimport static org.hawkular.metrics.core.api.MetricType.COUNTER;%0Aimport static org.hawkular.metrics.core.api.MetricType.COUNTER_RATE;%0A%0A
 /**%0A
@@ -2730,641 +2730,204 @@
     
-logger.debug(%22start = %7B%7D, end = %7B%7D%22, start, end);%0A        metricsService.findCounterData(id, start, end)%0A                .take(1)%0A                .doOnNext(dataPoint -%3E logger.debug(%22Data Point = %7B%7D%22, dataPoint))%0A                .map(dataPoint -%3E ((dataPoint.getValue().doubleValue() / (end - start) * 1000)))%0A                .map(rate -%3E new Metric%3C%3E(new MetricId(id.getTenantId(), COUNTER_RATE, id.getName()),%0A                        singletonList(new DataPoint%3C%3E(start, rate))))%0A                .flatMap(metric -%3E metricsService.addGaugeData(Observable.just(metric)))%0A                .subscribe(%0A                        aVoid -%3E %7B
+updates.subscribe(%0A                aVoid -%3E %7B%0A                %7D,%0A                t -%3E %7B%0A                    logger.warn(%22There was an error persisting rates for %7Btenant= %22 + tenant + %22, start= %22 +
 %0A   
@@ -2951,125 +2951,47 @@
     
-%7D,%0A
     
-                    t -%3E %7B%0A                            logger.warn(%22Failed to persist rate data%22, t);%0A        
+start + %22, end= %22 + end + %22%7D%22, t);%0A
     
@@ -3037,39 +3037,27 @@
             
-        %7D,%0A    
+%7D,%0A
             
@@ -3064,20 +3064,16 @@
     
-    
 () -%3E %7B%0A
     
@@ -3064,32 +3064,24 @@
     () -%3E %7B%0A
-        
             
@@ -3134,28 +3134,117 @@
 ate data
-%22);%0A        
+ for %7Btenant= %22 + tenant + %22, start= %22 +%0A                            start + %22, end= %22 + end + %22%7D%22);%0A
         
@@ -3290,24 +3290,26 @@
         
+%7D%0A
         
 %7D%0A      
@@ -3292,33 +3292,35 @@
       %7D%0A        
-%7D
+);%0A
 %0A               
@@ -3312,26 +3312,100 @@
         
-        );
+// TODO We do not want to block but have to for now. See HWKMETRICS-214 for details.
 %0A       

add some comments