package org.coursera.metrics.datadog.transport;

import com.timgroup.statsd.NonBlockingStatsDClient;
import com.timgroup.statsd.StatsDClient;
import org.coursera.metrics.datadog.model.DatadogCounter;
import org.coursera.metrics.datadog.model.DatadogGauge;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * Uses dogstatsd UDP protocol to push metrics to datadog. Note that datadog doesn't support
 * time in the UDP protocol. So all metrics are against current time.
 * <p/>
 * Also dogstatsd UDP doesn't support batching of metrics, so it pushes metrics as it receives
 * rather than batching.
 *
 * @see <a href="http://docs.datadoghq.com/guides/dogstatsd">dogstatsd</a>
 */
public class UdpTransport implements Transport {

  private static final Logger LOG = LoggerFactory.getLogger(UdpTransport.class);
  private final StatsDClient statsd;

  private UdpTransport(String prefix, String statsdHost, int port, String[] globalTags) {
    statsd = new NonBlockingStatsDClient(
        prefix,
        statsdHost,
        port,
        globalTags
    );
  }

  public void close() throws IOException {
    statsd.stop();
  }

  public static class Builder {
    String prefix = null;
    String statsdHost = "localhost";
    int port = 8125;

    public Builder withPrefix(String prefix) {
      this.prefix = prefix;
      return this;
    }

    public Builder withStatsdHost(String statsdHost) {
      this.statsdHost = statsdHost;
      return this;
    }

    public Builder withPort(int port) {
      this.port = port;
      return this;
    }

    public UdpTransport build() {
      return new UdpTransport(prefix, statsdHost, port, new String[0]);
    }
  }

  public Request prepare() throws IOException {
    return new DogstatsdRequest(statsd);
  }

  public static class DogstatsdRequest implements Transport.Request {
    private final StatsDClient statsdClient;

    public DogstatsdRequest(StatsDClient statsdClient) {
      this.statsdClient = statsdClient;
    }

    /**
     * statsd has no notion of batch request, so gauges are pushed as they are received
     */
    public void addGauge(DatadogGauge gauge) {
      if (gauge.getPoints().size() > 1) {
        LOG.debug("Gauge " + gauge.getMetric() + " has more than one data point, " +
            "will pick the first point only");
      }
      double value = gauge.getPoints().get(0).get(1).doubleValue();
      String[] tags = gauge.getTags().toArray(new String[gauge.getTags().size()]);
      statsdClient.gauge(gauge.getMetric(), value, tags);
    }

    /**
     * statsd has no notion of batch request, so counters are pushed as they are received
     */
    public void addCounter(DatadogCounter counter) {
      if (counter.getPoints().size() > 1) {
        LOG.debug("Counter " + counter.getMetric() + " has more than one data point, " +
            "will pick the first point only");
      }
      int value = counter.getPoints().get(0).get(1).intValue();
      String[] tags = counter.getTags().toArray(new String[counter.getTags().size()]);
      statsdClient.count(counter.getMetric(), value, tags);
    }

    /**
     * For statsd the metrics are pushed as they are received. So there is nothing do in send.
     */
    public void send() {
    }
  }
}

@@ -334,16 +334,64 @@
 ception;
+%0Aimport java.util.HashMap;%0Aimport java.util.Map;
 %0A%0A/**%0A *
@@ -928,16 +928,90 @@
  statsd;
+%0A  private final HashMap lastSeenCounters = new HashMap%3CString,Integer%3E();
 %0A%0A  priv
@@ -1912,16 +1912,34 @@
 t(statsd
+, lastSeenCounters
 );%0A  %7D%0A%0A
@@ -2052,16 +2052,72 @@
 dClient;
+%0A    private final Map%3CString,Integer%3E lastSeenCounters;
 %0A%0A    pu
@@ -2163,16 +2163,54 @@
 sdClient
+, Map%3CString,Integer%3E lastSeenCounters
 ) %7B%0A    
@@ -2241,24 +2241,72 @@
 atsdClient;%0A
+      this.lastSeenCounters = lastSeenCounters;%0A
     %7D%0A%0A    /
@@ -3351,55 +3351,588 @@
 %5D);%0A
+%0A
       
-statsdClient.count(c
+String metric = counter.getMetric();%0A      int finalValue = value;%0A      if (lastSeenCounters.containsKey(metric)) %7B%0A        // If we've seen this counter before then calculate the difference%0A        // by subtracting the new value from the old. StatsD expects a relative%0A        // counter, not an absolute!%0A        finalValue = value - lastSeenC
 ounter
+s
 .get
-M
+(m
 etric
-(), v
+);%0A      %7D%0A      // Store the last value we saw so that the next addCounter call can make%0A      // the proper relative value%0A      lastSeenCounters.put(metric, value);%0A%0A      statsdClient.count(metric, finalV
 alue

Make UDP counters relative.