test

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

@@ -3,6 +3,7 @@
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 import java.nio.channels.DatagramChannel;
 import java.nio.charset.Charset;
@@ -11,6 +12,7 @@
 import java.text.NumberFormat;
 import java.util.Locale;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -118,7 +120,7 @@ protected NumberFormat initialValue() {
      *     if the client could not be started
      */
     public NonBlockingStatsDClient(String prefix, String hostname, int port) throws StatsDClientException {
-        this(prefix, hostname, port, null, NO_OP_HANDLER);
+        this(prefix, hostname, port, null, null);
     }
 
     /**
@@ -143,7 +145,7 @@ public NonBlockingStatsDClient(String prefix, String hostname, int port) throws
      *     if the client could not be started
      */
     public NonBlockingStatsDClient(String prefix, String hostname, int port, String... constantTags) throws StatsDClientException {
-        this(prefix, hostname, port, constantTags, NO_OP_HANDLER);
+        this(prefix, hostname, port, constantTags, null);
     }
 
     /**
@@ -166,17 +168,48 @@ public NonBlockingStatsDClient(String prefix, String hostname, int port, String.
      * @param constantTags
      *     tags to be added to all content sent
      * @param errorHandler
-     *     handler to use when an exception occurs during usage
+     *     handler to use when an exception occurs during usage, may be null to indicate noop
      * @throws StatsDClientException
      *     if the client could not be started
      */
-    public NonBlockingStatsDClient(String prefix, String hostname, int port, String[] constantTags, StatsDClientErrorHandler errorHandler) throws StatsDClientException {
+    public NonBlockingStatsDClient(String prefix, final String hostname, final int port, String[] constantTags, StatsDClientErrorHandler errorHandler) throws StatsDClientException {
+        this(prefix, constantTags, errorHandler, staticStatsDAddressResolution(hostname, port));
+    }
+
+    /**
+     * Create a new StatsD client communicating with a StatsD instance on the
+     * specified host and port. All messages send via this client will have
+     * their keys prefixed with the specified string. The new client will
+     * attempt to open a connection to the StatsD server immediately upon
+     * instantiation, and may throw an exception if that a connection cannot
+     * be established. Once a client has been instantiated in this way, all
+     * exceptions thrown during subsequent usage are passed to the specified
+     * handler and then consumed, guaranteeing that failures in metrics will
+     * not affect normal code execution.
+     *
+     * @param prefix
+     *     the prefix to apply to keys sent via this client
+     * @param constantTags
+     *     tags to be added to all content sent
+     * @param errorHandler
+     *     handler to use when an exception occurs during usage, may be null to indicate noop
+     * @param addressLookup
+     *     yields the IP address and socket of the StatsD server
+     * @throws StatsDClientException
+     *     if the client could not be started
+     */
+    public NonBlockingStatsDClient(String prefix, String[] constantTags, StatsDClientErrorHandler errorHandler, Callable<InetSocketAddress> addressLookup) throws StatsDClientException {
         if(prefix != null && prefix.length() > 0) {
             this.prefix = String.format("%s.", prefix);
         } else {
             this.prefix = "";
         }
-        this.handler = errorHandler;
+        if(errorHandler == null) {
+            this.handler = NO_OP_HANDLER;
+        }
+        else {
+            this.handler = errorHandler;
+        }
 
         /* Empty list should be null for faster comparison */
         if(constantTags != null && constantTags.length == 0) {
@@ -194,7 +227,7 @@ public NonBlockingStatsDClient(String prefix, String hostname, int port, String[
         } catch (Exception e) {
             throw new StatsDClientException("Failed to start StatsD client", e);
         }
-        this.executor.submit(new QueueConsumer(hostname,port));
+        this.executor.submit(new QueueConsumer(addressLookup));
     }
 
     /**
@@ -552,29 +585,28 @@ private void send(String message) {
     private class QueueConsumer implements Runnable {
         private final ByteBuffer sendBuffer = ByteBuffer.allocate(PACKET_SIZE_BYTES);
 
-        private String hostname;
-        private int port;
+        private final Callable<InetSocketAddress> addressLookup;
 
-        public QueueConsumer(String hostname, int port) {
-            this.hostname = hostname;
-            this.port = port;
+        public QueueConsumer(Callable<InetSocketAddress> addressLookup) {
+            this.addressLookup = addressLookup;
         }
 
         @Override public void run() {
             while(!executor.isShutdown()) {
                 try {
                     String message = queue.poll(1, TimeUnit.SECONDS);
                     if(null != message) {
+                        InetSocketAddress address = this.addressLookup.call();
                         byte[] data = message.getBytes(MESSAGE_CHARSET);
                         if(sendBuffer.remaining() < (data.length + 1)) {
-                            blockingSend();
+                            blockingSend(address);
                         }
                         if(sendBuffer.position() > 0) {
                             sendBuffer.put( (byte) '\n');
                         }
                         sendBuffer.put(data);
                         if(null == queue.peek()) {
-                            blockingSend();
+                            blockingSend(address);
                         }
                     }
                 } catch (Exception e) {
@@ -583,12 +615,10 @@ public QueueConsumer(String hostname, int port) {
             }
         }
 
-        private void blockingSend() throws IOException {
+        private void blockingSend(InetSocketAddress address) throws IOException {
             int sizeOfBuffer = sendBuffer.position();
             sendBuffer.flip();
 
-            InetSocketAddress address = new InetSocketAddress(InetAddress.getByName(this.hostname),this.port);
-
             int sentBytes = clientChannel.send(sendBuffer, address);
             sendBuffer.limit(sendBuffer.capacity());
             sendBuffer.rewind();
@@ -606,4 +636,45 @@ private void blockingSend() throws IOException {
             }
         }
     }
+
+    /**
+     * Create dynamic lookup for the given host name and port.
+     *
+     * @param hostname the host name of the targeted StatsD server
+     * @param port     the port of the targeted StatsD server
+     * @return a function to perform the lookup
+     * @see NonBlockingStatsDClient#NonBlockingStatsDClient(String, String[], StatsDClientErrorHandler, Callable)
+     */
+    public static Callable<InetSocketAddress> volatileAddressResolution(final String hostname, final int port) {
+        return new Callable<InetSocketAddress>() {
+            @Override public InetSocketAddress call() throws Exception {
+                return new InetSocketAddress(InetAddress.getByName(hostname), port);
+            }
+        };
+    }
+
+  /**
+   * Lookup the address for the given host name and cache the result.
+   *
+   * @param hostname the host name of the targeted StatsD server
+   * @param port     the port of the targeted StatsD server
+   * @return a function that cached the result of the lookup
+   * @throws Exception if the lookup fails, i.e. {@link UnknownHostException}
+   */
+    public static Callable<InetSocketAddress> staticAddressResolution(String hostname, int port) throws Exception {
+        final InetSocketAddress address = volatileAddressResolution(hostname, port).call();
+        return new Callable<InetSocketAddress>() {
+            @Override public InetSocketAddress call() throws Exception {
+                return address;
+            }
+        };
+    }
+
+    private static Callable<InetSocketAddress> staticStatsDAddressResolution(String hostname, int port) throws StatsDClientException {
+        try {
+            return staticAddressResolution(hostname, port);
+        } catch (Exception e) {
+            throw new StatsDClientException("Failed to lookup StatsD host", e);
+        }
+    }
 }