package net.dempsy.monitoring.dropwizard;

import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.Timer;

import net.dempsy.config.ClusterId;
import net.dempsy.monitoring.ClusterStatsCollector;
import net.dempsy.monitoring.StatsCollector;
import net.dempsy.utils.MetricUtils;

public class DropwizardClusterStatsCollector implements ClusterStatsCollector {

    public static final String MESSAGES_DISPATCHED = "messages-dispatched";
    public static final String MESSAGES_PROCESSED = "messages-processed";
    public static final String MESSAGES_FAILED = "messages-failed";
    public static final String MESSAGES_COLLISION = "messages-collision";
    public static final String MESSAGES_DISCARDED = "messages-discarded";
    public static final String MESSAGES_PROCESSOR_CREATED = "messages-processor-created";
    public static final String MESSAGES_PROCESSOR_DELETED = "messages-processor-deleted";
    public static final String OUTPUT_INVOKE_STARTED_TIMER = "output-invoke-started-timer";
    public static final String EVICTION_PASS_STARTED_TIMER = "eviction-pass-started-timer";
    public static final String PRE_INSTANTIATION_STARTED_TIMER = "pre-instantiation-started-timer";

    public static final String[] METRIC_NAMES = new String[] {
        MESSAGES_DISPATCHED,
        MESSAGES_PROCESSED,
        MESSAGES_FAILED,
        MESSAGES_COLLISION,
        MESSAGES_DISCARDED,
        MESSAGES_PROCESSOR_CREATED,
        MESSAGES_PROCESSOR_DELETED,
        OUTPUT_INVOKE_STARTED_TIMER,
        EVICTION_PASS_STARTED_TIMER,
        PRE_INSTANTIATION_STARTED_TIMER
    };

    private static class DropwizardTimerContext implements StatsCollector.TimerContext {
        private final Timer timer;
        private final Timer.Context context;

        private DropwizardTimerContext(final MetricRegistry registry, final String timerName) {
            timer = registry.timer(timerName);
            context = timer.time();
        }

        @Override
        public void stop() {
            context.stop();
        }
    }

    private final ClusterId clusterId;

    private final MetricRegistry registry;
    private final Meter messageDispatched;
    private final Meter messageProcessed;
    private final Meter messageFailed;
    private final Meter messageCollision;
    private final Meter messageDiscarded;
    private final Meter messageProcessorCreated;
    private final Meter messageProcessorDeleted;

    public DropwizardClusterStatsCollector(final ClusterId clusterId) {
        super();

        this.clusterId = clusterId;

        registry = MetricUtils.getMetricsRegistry();
        messageDispatched = registry.meter(getName(MESSAGES_DISPATCHED));
        messageProcessed = registry.meter(getName(MESSAGES_PROCESSED));
        messageFailed = registry.meter(getName(MESSAGES_FAILED));
        messageCollision = registry.meter(getName(MESSAGES_COLLISION));
        messageDiscarded = registry.meter(getName(MESSAGES_DISCARDED));
        messageProcessorCreated = registry.meter(getName(MESSAGES_PROCESSOR_CREATED));
        messageProcessorDeleted = registry.meter(getName(MESSAGES_PROCESSOR_DELETED));
    }

    @Override
    public void messageDispatched(final int num) {
        messageDispatched.mark(num);
    }

    @Override
    public void messageProcessed(final int num) {
        messageProcessed.mark(num);
    }

    @Override
    public void messageFailed(final int num) {
        messageFailed.mark(num);
    }

    @Override
    public void messageCollision(final Object message) {
        messageCollision.mark();
    }

    @Override
    public void messageDiscarded(final Object message) {
        messageDiscarded.mark();
    }

    @Override
    public void messageProcessorCreated(final Object key) {
        messageProcessorCreated.mark();
    }

    @Override
    public void messageProcessorDeleted(final Object key) {
        messageProcessorDeleted.mark();
    }

    @Override
    public void stop() {
        // Remove the metrics from the registry
        for(final String m: METRIC_NAMES) {
            registry.remove(getName(m));
        }
    }

    @Override
    public TimerContext preInstantiationStarted() {
        return new DropwizardTimerContext(registry, getName(PRE_INSTANTIATION_STARTED_TIMER));
    }

    @Override
    public TimerContext outputInvokeStarted() {
        return new DropwizardTimerContext(registry, getName(OUTPUT_INVOKE_STARTED_TIMER));
    }

    @Override
    public TimerContext evictionPassStarted() {
        return new DropwizardTimerContext(registry, getName(EVICTION_PASS_STARTED_TIMER));
    }

    // protected access for testing purposes
    protected String getName(final String key) {
        return MetricRegistry.name(DropwizardClusterStatsCollector.class, "cluster", clusterId.applicationName, clusterId.clusterName, key);
    }
}

@@ -36,16 +36,64 @@
 izard;%0A%0A
+import java.util.concurrent.atomic.AtomicLong;%0A%0A
 import c
@@ -119,16 +119,16 @@
 .Meter;%0A
-
 import c
@@ -813,24 +813,94 @@
 discarded%22;%0A
+    public static final String MESSAGES_PENDING = %22messages-pending%22;%0A
     public s
@@ -1738,16 +1738,42 @@
 ED_TIMER
+,%0A        MESSAGES_PENDING
 %0A    %7D;%0A
@@ -2611,24 +2611,92 @@
 orDeleted;%0A%0A
+    private final AtomicLong inProcessMessages = new AtomicLong();%0A%0A
     public D
@@ -3389,24 +3389,112 @@
 _DELETED));%0A
+        registry.gauge(getName(MESSAGES_PENDING), () -%3E () -%3E inProcessMessages.get());%0A
     %7D%0A%0A    @
@@ -3582,32 +3582,74 @@
 ched.mark(num);%0A
+        inProcessMessages.getAndAdd(num);%0A
     %7D%0A%0A    @Over
@@ -3731,32 +3731,75 @@
 ssed.mark(num);%0A
+        inProcessMessages.getAndAdd(-num);%0A
     %7D%0A%0A    @Over
@@ -3842,32 +3842,32 @@
 inal int num) %7B%0A
-
         messageF
@@ -3879,24 +3879,67 @@
 .mark(num);%0A
+        inProcessMessages.getAndAdd(-num);%0A
     %7D%0A%0A    @

Add a messages pending guage to the monitored metrics.