package org.atlasapi.messaging;

import static com.google.common.base.Preconditions.checkNotNull;

import javax.annotation.Nullable;

import org.atlasapi.entity.util.WriteException;
import org.atlasapi.equivalence.EquivalenceGraphStore;

import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.Timer;
import com.metabroadcast.common.queue.Worker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ContentEquivalenceUpdatingWorker implements Worker<EquivalenceAssertionMessage> {

    private final Logger log = LoggerFactory.getLogger(ContentEquivalenceUpdatingWorker.class);

    private final EquivalenceGraphStore graphStore;
    @Nullable private final Timer messageTimer;

    public ContentEquivalenceUpdatingWorker(EquivalenceGraphStore graphStore, @Nullable
    MetricRegistry metrics) {
        this.graphStore = checkNotNull(graphStore);
        this.messageTimer = (metrics != null ?
                             checkNotNull(metrics.timer("ContentEquivalenceUpdatingWorker")) :
                             null);
    }

    @Override
    public void process(EquivalenceAssertionMessage message) {
        try {
            if (messageTimer != null) {
                Timer.Context timer = messageTimer.time();
                graphStore.updateEquivalences(message.getSubject(), message.getAssertedAdjacents(),
                        message.getPublishers());
                timer.stop();
            } else {
                graphStore.updateEquivalences(message.getSubject(), message.getAssertedAdjacents(),
                        message.getPublishers());
            }
        } catch (Exception e) {
            log.warn("Error while processing EquivalenceAssertionMessage {}", message.toString(), e);
        }
    }

}

@@ -138,47 +138,34 @@
 ort 
-org.atlasapi.entity.util.WriteException
+com.codahale.metrics.Meter
 ;%0Aim
@@ -693,16 +693,57 @@
 geTimer;
+%0A    @Nullable private final Meter meter;
 %0A%0A    pu
@@ -1084,16 +1084,171 @@
  null);%0A
+        this.meter = (metrics != null ?%0A                checkNotNull(metrics.meter(%22ContentEquivalenceUpdatingWorker.errorRate%22)) :%0A                null);%0A
     %7D%0A%0A 
@@ -1797,24 +1797,104 @@
           %7D%0A
+            log.debug(%22Successfully processed message %7B%7D%22, message.toString());%0A
         %7D ca
@@ -2011,24 +2011,101 @@
 ring(), e);%0A
+            if (meter != null) %7B%0A                meter.mark();%0A            %7D%0A
         %7D%0A  

Add some more logging and error rate Metric.