package edu.brown.cs.systems.retro.aggregation.reporters;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import com.typesafe.config.Config;

import edu.brown.cs.systems.pubsub.PubSub;
import edu.brown.cs.systems.retro.aggregation.Report.ImmediateReport;
import edu.brown.cs.systems.retro.aggregation.Report.ResourceReport;
import edu.brown.cs.systems.retro.aggregation.Resource;
import edu.brown.cs.systems.retro.aggregation.aggregators.ResourceAggregator;

/**
 * Periodically sends resource reports
 * 
 * @author a-jomace
 */
public class PubSubReporter implements Runnable, Reporter {

    private final Config config;
    private final String immediateReportTopic;
    private final ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
    private final Map<ResourceAggregator, String> resources = new HashMap<ResourceAggregator, String>();

    public PubSubReporter(Config config) {
        this.config = config;
        this.immediateReportTopic = getImmediateReportTopic(config);

        // Get the reporting interval from the config
        long interval = config.getLong("resource-reporting.reporting.interval");

        // Determine the first reporting interval - as much as possible, align
        // with clock boundaries
        long initial_delay = 2 * interval - (System.currentTimeMillis() % interval);

        // Schedule at a fixed rate
        executor.scheduleAtFixedRate(this, initial_delay, interval, TimeUnit.MILLISECONDS);
        System.out.println("Resource reporting executor started");
    }

    @Override
    public synchronized void register(ResourceAggregator aggregator) {
        if (!resources.containsKey(aggregator)) {
            String topic = getTopic(config, aggregator.type);
            register(topic, aggregator);
        }
    }

    public synchronized void register(String topic, ResourceAggregator aggregator) {
        if (!resources.containsKey(aggregator)) {
            if (topic != null) {
                resources.put(aggregator, topic);
                System.out.println("ZmqReporter: " + aggregator + " -> " + topic);
            }
        }

    }

    public static String getTopic(Config config, Resource.Type type) {
        String suffix = "default";

        switch (type) {
        case DISK:
            suffix = "disk";
            break;
        case DISKCACHE:
            suffix = "disk-cache";
            break;
        case NETWORK:
            suffix = "network";
            break;
        case LOCKING:
            suffix = "locks";
            break;
        case CPU:
            suffix = "cpu";
            break;
        case HDFSREQUEST:
            suffix = "hdfs";
            break;
        case QUEUE:
            suffix = "queue";
            break;
        case THROTTLINGPOINT:
            suffix = "throttlingpoint";
            break;
        default:
            break;
        }

        String property = "resource-reporting.reporting.zmq.topics." + suffix;
        try {
            return config.getString(property);
        } catch (Exception e) {
            return null;
        }
    }

    public static String getImmediateReportTopic(Config config) {
        try {
            return config.getString("resource-reporting.reporting.zmq.topics.immediate");
        } catch (Exception e) {
            return null;
        }
    }

    @Override
    public final void run() {
        synchronized (this) {
            for (ResourceAggregator r : resources.keySet()) {
                if (r.enabled()) {
                    String topic = resources.get(r);
                    try {
                        ResourceReport report = r.getReport();
                        if (report != null)
                            PubSub.publish(topic, report);
                    } catch (Exception e) {
                        System.out.println("Exception sending resource report on topic " + topic);
                        System.out.println(r);
                        e.printStackTrace();
                    }
                }
            }
        }
    }

    /**
     * publishes the immediate report on the immediate report topic. Not
     * published inband, instead, immediately submitted to executor to build and
     * send the report
     */
    public void reportImmediately(final ImmediateReport.Builder report) {
        if (report != null && immediateReportTopic != null) {
            Runnable doReport = new Runnable() {
                public void run() {
                    try {
                        PubSub.publish(immediateReportTopic, report.build());
                    } catch (Exception e) {
                        System.out.println("Exception sending immediate report on topic " + immediateReportTopic);
                    }
                }
            };
            executor.submit(doReport);
        }
    }

}

@@ -225,18 +225,131 @@
 ent.
-TimeUnit;%0A
+ScheduledThreadPoolExecutor;%0Aimport java.util.concurrent.TimeUnit;%0A%0Aimport com.google.common.util.concurrent.MoreExecutors;
 %0Aimp
@@ -833,142 +833,248 @@
 r %7B%0A
+    
 %0A    
-private final Config config;%0A    private final String immediateReportTopic;%0A    private final ScheduledExecutorService e
+// Runs threads as daemon threads, but ensures runnables finish before exiting%0A    private static final ScheduledExecutorService executor = MoreExecutors.getExitingScheduledExecutorService(%0A            (ScheduledThreadPoolE
 xecutor
- =
+)
  Exe
@@ -1105,16 +1105,119 @@
 dPool(1)
+, 2, TimeUnit.SECONDS);%0A%0A    private final Config config;%0A    private final String immediateReportTopic
 ;%0A    pr
@@ -1984,24 +1984,130 @@
  started%22);%0A
+        %0A        Runtime.getRuntime().addShutdownHook(new Thread(this)); // Also publish once on shutdown%0A
     %7D%0A%0A    @

Make retro threads daemon threads