direct: fix sampling and visibility issues with new direct client

- sampling should always be disabled for distribution values with the direct client
  this is the whole point of being able to send multiple distribution values at once
- don't make the direct client final, users might need to have to extend it (same
  for the noop client)

src/main/java/com/timgroup/statsd/NoOpDirectStatsDClient.java

@@ -4,7 +4,7 @@
  * A No-Op {@link NonBlockingDirectStatsDClient}, which can be substituted in when metrics are not
  * required.
  */
-public final class NoOpDirectStatsDClient extends NoOpStatsDClient implements DirectStatsDClient {
+public class NoOpDirectStatsDClient extends NoOpStatsDClient implements DirectStatsDClient {
     @Override public void recordDistributionValues(String aspect, double[] values, double sampleRate, String... tags) { }
 
     @Override public void recordDistributionValues(String aspect, long[] values, double sampleRate, String... tags) { }

src/main/java/com/timgroup/statsd/NonBlockingDirectStatsDClient.java

@@ -1,32 +1,22 @@
 package com.timgroup.statsd;
 
-import static java.nio.charset.StandardCharsets.UTF_8;
-
-final class NonBlockingDirectStatsDClient extends NonBlockingStatsDClient implements DirectStatsDClient {
+class NonBlockingDirectStatsDClient extends NonBlockingStatsDClient implements DirectStatsDClient {
 
     public NonBlockingDirectStatsDClient(final NonBlockingStatsDClientBuilder builder) throws StatsDClientException {
         super(builder);
     }
 
     @Override
     public void recordDistributionValues(String aspect, double[] values, double sampleRate, String... tags) {
-        if ((Double.isNaN(sampleRate) || !isInvalidSample(sampleRate)) && values != null && values.length > 0) {
-            if (values.length == 1) {
-                recordDistributionValue(aspect, values[0], sampleRate, tags);
-            } else {
-                sendMetric(new DoublesStatsDMessage(aspect, Message.Type.DISTRIBUTION, values, sampleRate, 0, tags));
-            }
+        if (values != null && values.length > 0) {
+            sendMetric(new DoublesStatsDMessage(aspect, Message.Type.DISTRIBUTION, values, sampleRate, 0, tags));
         }
     }
 
     @Override
     public void recordDistributionValues(String aspect, long[] values, double sampleRate, String... tags) {
-        if ((Double.isNaN(sampleRate) || !isInvalidSample(sampleRate)) && values != null && values.length > 0) {
-            if (values.length == 1) {
-                recordDistributionValue(aspect, values[0], sampleRate, tags);
-            } else {
-                sendMetric(new LongsStatsDMessage(aspect, Message.Type.DISTRIBUTION, values, sampleRate, 0, tags));
-            }
+        if (values != null && values.length > 0) {
+            sendMetric(new LongsStatsDMessage(aspect, Message.Type.DISTRIBUTION, values, sampleRate, 0, tags));
         }
     }
 

src/test/java/com/timgroup/statsd/NonBlockingDirectStatsDClientTest.java

@@ -86,6 +86,14 @@ public void sends_multivalued_distribution_to_statsd_with_sampling_rate() {
         assertThat(server.messagesReceived(), hasItem(comparesEqualTo("my.prefix.mydistribution:423:234|d|@1.000000")));
     }
 
+    @Test(timeout = 5000L)
+    public void sends_multivalued_distribution_to_statsd_with_non_1_sampling_rate() {
+        client.recordDistributionValues("mydistribution", new long[] { 423L, 234L }, 0.1);
+        server.waitForMessage("my.prefix");
+
+        assertThat(server.messagesReceived(), hasItem(comparesEqualTo("my.prefix.mydistribution:423:234|d|@0.100000")));
+    }
+
     @Test(timeout = 5000L)
     public void sends_multivalued_distribution_to_statsd_with_tags_and_sampling_rate() {
         client.recordDistributionValues("mydistribution", new long[] { 423L, 234L }, 1, "foo:bar", "baz");
@@ -97,19 +105,19 @@ public void sends_multivalued_distribution_to_statsd_with_tags_and_sampling_rate
     @Test(timeout = 5000L)
     public void sends_too_long_multivalued_distribution_to_statsd() {
         long[] values = {423L, 234L, 456L, 512L, 345L, 898L, 959876543123L, 667L};
-        client.recordDistributionValues("mydistribution", values, 1, "foo:bar", "baz");
+        client.recordDistributionValues("mydistribution", values, 0.4, "foo:bar", "baz");
 
         server.waitForMessage("my.prefix");
-        assertThat(server.messagesReceived(), hasItem(comparesEqualTo("my.prefix.mydistribution:423:234:456|d|@1.000000|#baz,foo:bar")));
+        assertThat(server.messagesReceived(), hasItem(comparesEqualTo("my.prefix.mydistribution:423:234:456|d|@0.400000|#baz,foo:bar")));
 
         server.waitForMessage("my.prefix");
-        assertThat(server.messagesReceived(), hasItem(comparesEqualTo("my.prefix.mydistribution:512:345:898|d|@1.000000|#baz,foo:bar")));
+        assertThat(server.messagesReceived(), hasItem(comparesEqualTo("my.prefix.mydistribution:512:345:898|d|@0.400000|#baz,foo:bar")));
 
         server.waitForMessage("my.prefix");
-        assertThat(server.messagesReceived(), hasItem(comparesEqualTo("my.prefix.mydistribution:959876543123|d|@1.000000|#baz,foo:bar")));
+        assertThat(server.messagesReceived(), hasItem(comparesEqualTo("my.prefix.mydistribution:959876543123|d|@0.400000|#baz,foo:bar")));
 
         server.waitForMessage("my.prefix");
-        assertThat(server.messagesReceived(), hasItem(comparesEqualTo("my.prefix.mydistribution:667|d|@1.000000|#baz,foo:bar")));
+        assertThat(server.messagesReceived(), hasItem(comparesEqualTo("my.prefix.mydistribution:667|d|@0.400000|#baz,foo:bar")));
     }
 
 }