feat: Add an API to send multiple samples at once

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

@@ -0,0 +1,46 @@
+package com.timgroup.statsd;
+
+/**
+ * DirectStatsDClient is an experimental extension of {@link StatsDClient} that allows for direct access to some
+ * dogstatsd features.
+ *
+ * <p>It is not recommended to use this client in production. This client might allow you to take advantage of
+ * new features in the agent before they are released, but it might also break your application.
+ */
+public interface DirectStatsDClient extends StatsDClient {
+
+    /**
+     * Records values for the specified named distribution.
+     *
+     * <p>The method doesn't take care of breaking down the values array if it is too large. It's up to the caller to
+     * make sure the size is kept reasonable.</p>
+     *
+     * <p>This method is a DataDog extension, and may not work with other servers.</p>
+     *
+     * <p>This method is non-blocking and is guaranteed not to throw an exception.</p>
+     *
+     * @param aspect     the name of the distribution
+     * @param values     the values to be incorporated in the distribution
+     * @param sampleRate percentage of time metric to be sent
+     * @param tags       array of tags to be added to the data
+     */
+    void recordDistributionValues(String aspect, double[] values, double sampleRate, String... tags);
+
+
+    /**
+     * Records values for the specified named distribution.
+     *
+     * <p>The method doesn't take care of breaking down the values array if it is too large. It's up to the caller to
+     * make sure the size is kept reasonable.</p>
+     *
+     * <p>This method is a DataDog extension, and may not work with other servers.</p>
+     *
+     * <p>This method is non-blocking and is guaranteed not to throw an exception.</p>
+     *
+     * @param aspect     the name of the distribution
+     * @param values     the values to be incorporated in the distribution
+     * @param sampleRate percentage of time metric to be sent
+     * @param tags       array of tags to be added to the data
+     */
+    void recordDistributionValues(String aspect, long[] values, double sampleRate, String... tags);
+}

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

@@ -54,10 +54,13 @@ protected Message(String aspect, Message.Type type, String[] tags) {
      * Write this message to the provided {@link StringBuilder}. Will
      * be called from the sender threads.
      *
-     * @param builder
-     *     StringBuilder the text representation will be written to.
+     * @param builder     StringBuilder the text representation will be written to.
+     * @param capacity    The capacity of the send buffer.
+     * @param containerID The container ID to be appended to the message.
+     * @return boolean indicating whether the message was partially written to the builder.
+     *     If true, the method will be called again with the same arguments to continue writing.
      */
-    abstract void writeTo(StringBuilder builder, String containerID);
+    abstract boolean writeTo(StringBuilder builder, int capacity, String containerID);
 
     /**
      * Aggregate message.

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

@@ -0,0 +1,11 @@
+package com.timgroup.statsd;
+
+/**
+ * A No-Op {@link NonBlockingDirectStatsDClient}, which can be substituted in when metrics are not
+ * required.
+ */
+public final 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/NoOpStatsDClient.java

@@ -7,7 +7,10 @@
  * @author Tom Denley
  *
  */
-public final class NoOpStatsDClient implements StatsDClient {
+public class NoOpStatsDClient implements StatsDClient {
+
+    NoOpStatsDClient() {}
+
     @Override public void stop() { }
 
     @Override public void close() { }

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

@@ -0,0 +1,159 @@
+package com.timgroup.statsd;
+
+final 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));
+            }
+        }
+    }
+
+    @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));
+            }
+        }
+    }
+
+    abstract class MultiValuedStatsDMessage extends Message {
+        private final double sampleRate; // NaN for none
+        private final long timestamp; // zero for none
+        private int metadataSize = -1; // Cache the size of the metadata, -1 means not calculated yet
+        private int offset = 0; // The index of the first value that has not been written
+
+        MultiValuedStatsDMessage(String aspect, Message.Type type, String[] tags, double sampleRate, long timestamp) {
+            super(aspect, type, tags);
+            this.sampleRate = sampleRate;
+            this.timestamp = timestamp;
+        }
+
+        @Override
+        public final boolean canAggregate() {
+            return false;
+        }
+
+        @Override
+        public final void aggregate(Message message) {
+        }
+
+        @Override
+        public final boolean writeTo(StringBuilder builder, int capacity, String containerID) {
+            int metadataSize = metadataSize(builder, containerID);
+            writeHeadMetadata(builder);
+            boolean partialWrite = writeValuesTo(builder, capacity - metadataSize);
+            writeTailMetadata(builder, containerID);
+            return partialWrite;
+
+        }
+
+        private int metadataSize(StringBuilder builder, String containerID) {
+            if (metadataSize == -1) {
+                int previousLength = builder.length();
+                writeHeadMetadata(builder);
+                writeTailMetadata(builder, containerID);
+                metadataSize = builder.length() - previousLength;
+                builder.setLength(previousLength);
+            }
+            return metadataSize;
+        }
+
+        private void writeHeadMetadata(StringBuilder builder) {
+            builder.append(prefix).append(aspect);
+        }
+
+        private void writeTailMetadata(StringBuilder builder, String containerID) {
+            builder.append('|').append(type);
+            if (!Double.isNaN(sampleRate)) {
+                builder.append('|').append('@').append(format(SAMPLE_RATE_FORMATTER, sampleRate));
+            }
+            if (timestamp != 0) {
+                builder.append("|T").append(timestamp);
+            }
+            tagString(tags, builder);
+            if (containerID != null && !containerID.isEmpty()) {
+                builder.append("|c:").append(containerID);
+            }
+
+            builder.append('\n');
+        }
+
+        private boolean writeValuesTo(StringBuilder builder, int remainingCapacity) {
+            int maxLength = builder.length() + remainingCapacity;
+
+            // Add at least one value
+            builder.append(':');
+            writeValueTo(builder, offset);
+            int previousLength = builder.length();
+
+            // Add remaining values up to the max length
+            for (int i = offset + 1; i < lengthOfValues(); i++) {
+                builder.append(':');
+                writeValueTo(builder, i);
+                if (builder.length() > maxLength) {
+                    builder.setLength(previousLength);
+                    offset += i;
+                    return true;
+                }
+                previousLength = builder.length();
+            }
+            offset = lengthOfValues();
+            return false;
+        }
+
+        protected abstract int lengthOfValues();
+
+        protected abstract void writeValueTo(StringBuilder buffer, int index);
+    }
+
+    final class LongsStatsDMessage extends MultiValuedStatsDMessage {
+        private final long[] values;
+
+        LongsStatsDMessage(String aspect, Message.Type type, long[] values, double sampleRate, long timestamp, String[] tags) {
+            super(aspect, type, tags, sampleRate, timestamp);
+            this.values = values;
+        }
+
+        @Override
+        protected int lengthOfValues() {
+            return values.length;
+        }
+
+        @Override
+        protected void writeValueTo(StringBuilder buffer, int index) {
+            buffer.append(values[index]);
+        }
+    }
+
+    final class DoublesStatsDMessage extends MultiValuedStatsDMessage {
+        private final double[] values;
+
+        DoublesStatsDMessage(String aspect, Message.Type type, double[] values, double sampleRate, long timestamp,
+                             String[] tags) {
+            super(aspect, type, tags, sampleRate, timestamp);
+            this.values = values;
+        }
+
+        @Override
+        protected int lengthOfValues() {
+            return values.length;
+        }
+
+        @Override
+        protected void writeValueTo(StringBuilder buffer, int index) {
+            buffer.append(values[index]);
+        }
+    }
+}

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

@@ -161,7 +161,7 @@ protected static String format(ThreadLocal<NumberFormat> formatter, Number value
         return formatter.get().format(value);
     }
 
-    private final String prefix;
+    final String prefix;
     private final ClientChannel clientChannel;
     private final ClientChannel telemetryClientChannel;
     private final StatsDClientErrorHandler handler;
@@ -247,7 +247,7 @@ protected static String format(ThreadLocal<NumberFormat> formatter, Number value
      * @throws StatsDClientException
      *     if the client could not be started
      */
-    private NonBlockingStatsDClient(final String prefix, final int queueSize, final String[] constantTags,
+    NonBlockingStatsDClient(final String prefix, final int queueSize, final String[] constantTags,
             final StatsDClientErrorHandler errorHandler, final Callable<SocketAddress> addressLookup,
             final Callable<SocketAddress> telemetryAddressLookup, final int timeout, final int bufferSize,
             final int maxPacketSizeBytes, String entityID, final int poolSize, final int processorWorkers,
@@ -526,7 +526,7 @@ protected StatsDMessage(String aspect, Message.Type type, T value, double sample
         }
 
         @Override
-        public final void writeTo(StringBuilder builder, String containerID) {
+        public final boolean writeTo(StringBuilder builder, int capacity, String containerID) {
             builder.append(prefix).append(aspect).append(':');
             writeValue(builder);
             builder.append('|').append(type);
@@ -542,6 +542,7 @@ public final void writeTo(StringBuilder builder, String containerID) {
             }
 
             builder.append('\n');
+            return false;
         }
 
         @Override
@@ -554,7 +555,7 @@ public boolean canAggregate() {
     }
 
 
-    private boolean sendMetric(final Message message) {
+    boolean sendMetric(final Message message) {
         return send(message);
     }
 
@@ -1145,7 +1146,7 @@ private StringBuilder eventMap(final Event event, StringBuilder res) {
     @Override
     public void recordEvent(final Event event, final String... eventTags) {
         statsDProcessor.send(new AlphaNumericMessage(Message.Type.EVENT, "") {
-            @Override public void writeTo(StringBuilder builder, String containerID) {
+            @Override public boolean writeTo(StringBuilder builder, int capacity, String containerID) {
                 final String title = escapeEventString(prefix + event.getTitle());
                 final String text = escapeEventString(event.getText());
                 builder.append(Message.Type.EVENT.toString())
@@ -1168,6 +1169,7 @@ public void recordEvent(final Event event, final String... eventTags) {
                 }
 
                 builder.append('\n');
+                return false;
             }
         });
         this.telemetry.incrEventsSent(1);
@@ -1200,13 +1202,14 @@ private int getUtf8Length(final String text) {
     @Override
     public void recordServiceCheckRun(final ServiceCheck sc) {
         statsDProcessor.send(new AlphaNumericMessage(Message.Type.SERVICE_CHECK, "") {
-            @Override public void writeTo(StringBuilder sb, String containerID) {
+            @Override
+            public boolean writeTo(StringBuilder sb, int capacity, String containerID) {
                 // see http://docs.datadoghq.com/guides/dogstatsd/#service-checks
                 sb.append(Message.Type.SERVICE_CHECK.toString())
-                    .append("|")
-                    .append(sc.getName())
-                    .append("|")
-                    .append(sc.getStatus());
+                        .append("|")
+                        .append(sc.getName())
+                        .append("|")
+                        .append(sc.getStatus());
                 if (sc.getTimestamp() > 0) {
                     sb.append("|d:").append(sc.getTimestamp());
                 }
@@ -1222,6 +1225,7 @@ public void recordServiceCheckRun(final ServiceCheck sc) {
                 }
 
                 sb.append('\n');
+                return false;
             }
         });
         this.telemetry.incrServiceChecksSent(1);
@@ -1286,7 +1290,8 @@ protected void writeValue(StringBuilder builder) {
                 builder.append(getValue());
             }
 
-            @Override protected final void writeTo(StringBuilder builder, String containerID) {
+            @Override
+            protected final boolean writeTo(StringBuilder builder, int capacity, String containerID) {
                 builder.append(prefix).append(aspect).append(':');
                 writeValue(builder);
                 builder.append('|').append(type);
@@ -1296,6 +1301,7 @@ protected void writeValue(StringBuilder builder) {
                 }
 
                 builder.append('\n');
+                return false;
             }
         });
     }

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

@@ -211,6 +211,18 @@ public NonBlockingStatsDClient build() throws StatsDClientException {
         return new NonBlockingStatsDClient(resolve());
     }
 
+    /**
+     * {@link DirectStatsDClient} factory method.
+     *
+     * <p>It is an experimental extension of {@link StatsDClient} that allows for direct access to some dogstatsd features.
+     * It is not recommended to use this client in production.
+     * @return the built DirectStatsDClient.
+     * @see DirectStatsDClient
+     */
+    public DirectStatsDClient buildDirectStatsDClient() throws StatsDClientException {
+        return new NonBlockingDirectStatsDClient(resolve());
+    }
+
     /**
      * Creates a copy of this builder with any implicit elements resolved.
      * @return the resolved copy of the builder.