/*
 * Copyright 2013 Rackspace
 *
 *    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 com.rackspacecloud.blueflood.inputs.processors;

import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.Futures;
import com.rackspacecloud.blueflood.concurrent.AsyncFunctionWithThreadPool;
import com.rackspacecloud.blueflood.concurrent.NoOpFuture;
import com.rackspacecloud.blueflood.io.DiscoveryIO;
import com.rackspacecloud.blueflood.types.Metric;

import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Counter;
import com.yammer.metrics.core.Meter;
import com.yammer.metrics.core.Timer;
import com.yammer.metrics.core.TimerContext;

import javax.security.auth.login.Configuration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadPoolExecutor;

public class DiscoveryWriter extends AsyncFunctionWithThreadPool<List<List<Metric>>, List<List<Metric>>> {

    private final List<DiscoveryIO> discoveryIOs = new ArrayList<DiscoveryIO>();
    private final Map<Class<? extends DiscoveryIO>, Timer> writeDurationTimers= new HashMap<Class<? extends DiscoveryIO>, Timer>();

    public DiscoveryWriter(ThreadPoolExecutor threadPool) {
        super(threadPool);
    }

    public void registerIO(DiscoveryIO io) {
        discoveryIOs.add(io);
        writeDurationTimers.put(io.getClass(),
                Metrics.newTimer(io.getClass(), "DiscoveryWriter Write Duration", TimeUnit.MILLISECONDS, TimeUnit.SECONDS)
                );
    }

    public ListenableFuture<List<Boolean>> processMetrics(List<List<Metric>> input) {
        final List<ListenableFuture<Boolean>> resultFutures = new ArrayList<ListenableFuture<Boolean>>();
        for (final List<Metric> metrics : input) {
            ListenableFuture<Boolean> futureBatchResult = getThreadPool().submit(new Callable<Boolean>() {
                public Boolean call() throws Exception {
                    boolean success = true;
                        for (DiscoveryIO io : discoveryIOs) {
                            TimerContext actualWriteCtx = writeDurationTimers.get(io.getClass()).time();
                            try {
                                io.insertDiscovery(metrics);
                            } catch (Exception ex) {
                                getLogger().error(ex.getMessage(), ex);
                                success = false;
                            }
                            // should we time failed attempts?
                            actualWriteCtx.stop();
                        }
                        return success;
                }
            });
            resultFutures.add(futureBatchResult);
        }
        return Futures.allAsList(resultFutures);
    }

    public ListenableFuture<List<List<Metric>>> apply(List<List<Metric>> input) {
        processMetrics(input);
        // we don't need all metrics to finish being inserted into the discovery backend
        // before moving onto the next step in the processing chain.
        return new NoOpFuture<List<List<Metric>>>(input);
    }
}

@@ -967,16 +967,72 @@
 veryIO;%0A
+import com.rackspacecloud.blueflood.service.CoreConfig;%0A
 import c
@@ -1073,16 +1073,16 @@
 Metric;%0A
-
 %0Aimport 
@@ -1269,16 +1269,73 @@
 Context;
+%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;
 %0A%0Aimport
@@ -1920,16 +1920,102 @@
 imer%3E();
+%0A    private static final Logger log = LoggerFactory.getLogger(DiscoveryWriter.class);
 %0A%0A    pu
@@ -2095,16 +2095,45 @@
 dPool);%0A
+        registerIOModules();%0A
     %7D%0A%0A 
@@ -2376,16 +2376,16 @@
 ECONDS)%0A
-
         
@@ -2394,32 +2394,1602 @@
       );%0A    %7D%0A%0A
+    public void registerIOModules() %7B%0A        List%3CString%3E modules = Configuration.getInstance().getListProperty(CoreConfig.DISCOVERY_MODULES);%0A%0A        ClassLoader classLoader = DiscoveryIO.class.getClassLoader();%0A        for (String module : modules) %7B%0A            log.info(%22Loading metric discovery module %22 + module);%0A            try %7B%0A                Class discoveryClass = classLoader.loadClass(module);%0A                DiscoveryIO discoveryIOModule = (DiscoveryIO) discoveryClass.newInstance();%0A                log.info(%22Registering metric discovery module %22 + module);%0A                registerIO(discoveryIOModule);%0A            %7D catch (InstantiationException e) %7B%0A                log.error(%22Unable to create instance of metric discovery class for: %22 + module, e);%0A                System.exit(1); //something better%0A            %7D catch (IllegalAccessException e) %7B%0A                log.error(%22Error starting metric discovery module: %22 + module, e);%0A                System.exit(1); //something better%0A            %7D catch (ClassNotFoundException e) %7B%0A                log.error(%22Unable to locate metric discovery module: %22 + module, e);%0A                System.exit(1); //something better%0A            %7D catch (RuntimeException e) %7B%0A                log.error(%22Error starting metric discovery module: %22 + module, e);%0A                System.exit(1); //something better%0A            %7D catch (Throwable e) %7B%0A                log.error(%22Error starting metric discovery module: %22 + module, e);%0A                System.exit(1); //something better%0A            %7D%0A        %7D%0A    %7D%0A%0A
     public Liste

Add method to load all configured discovery modules