diff --git a/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java b/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java index 4ea7591c..eacaba62 100644 --- a/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java +++ b/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java @@ -96,6 +96,7 @@ String tag() { public static final int SOCKET_BUFFER_BYTES = -1; public static final boolean DEFAULT_BLOCKING = false; public static final boolean DEFAULT_ENABLE_TELEMETRY = true; + public static final boolean DEFAULT_ENABLE_JDK_SOCKET = true; public static final boolean DEFAULT_ENABLE_AGGREGATION = true; public static final boolean DEFAULT_ENABLE_ORIGIN_DETECTION = true; @@ -248,7 +249,8 @@ public NonBlockingStatsDClient(final NonBlockingStatsDClientBuilder builder) builder.addressLookup, builder.timeout, builder.connectionTimeout, - builder.socketBufferSize); + builder.socketBufferSize, + builder.enableJdkSocket); ThreadFactory threadFactory = builder.threadFactory != null @@ -296,7 +298,8 @@ public NonBlockingStatsDClient(final NonBlockingStatsDClientBuilder builder) builder.telemetryAddressLookup, builder.timeout, builder.connectionTimeout, - builder.socketBufferSize); + builder.socketBufferSize, + builder.enableJdkSocket); // similar settings, but a single worker and non-blocking. telemetryStatsDProcessor = @@ -482,7 +485,8 @@ ClientChannel createByteChannel( Callable addressLookup, int timeout, int connectionTimeout, - int bufferSize) + int bufferSize, + boolean enableJdkSocket) throws Exception { final SocketAddress address = addressLookup.call(); if (address instanceof NamedPipeSocketAddress) { @@ -497,7 +501,11 @@ ClientChannel createByteChannel( switch (unixAddr.getTransportType()) { case UDS_STREAM: return new UnixStreamClientChannel( - unixAddr.getAddress(), timeout, connectionTimeout, bufferSize); + unixAddr.getAddress(), + timeout, + connectionTimeout, + bufferSize, + enableJdkSocket); case UDS_DATAGRAM: case UDS: return new UnixDatagramClientChannel( diff --git a/src/main/java/com/timgroup/statsd/NonBlockingStatsDClientBuilder.java b/src/main/java/com/timgroup/statsd/NonBlockingStatsDClientBuilder.java index 289b16c2..696523d3 100644 --- a/src/main/java/com/timgroup/statsd/NonBlockingStatsDClientBuilder.java +++ b/src/main/java/com/timgroup/statsd/NonBlockingStatsDClientBuilder.java @@ -52,6 +52,9 @@ public class NonBlockingStatsDClientBuilder implements Cloneable { public boolean enableAggregation = NonBlockingStatsDClient.DEFAULT_ENABLE_AGGREGATION; + /** Enable native JDK support for UDS. Only available on Java 16+. */ + public boolean enableJdkSocket = NonBlockingStatsDClient.DEFAULT_ENABLE_JDK_SOCKET; + /** Telemetry flush interval, in milliseconds. */ public int telemetryFlushInterval = Telemetry.DEFAULT_FLUSH_INTERVAL; @@ -322,6 +325,11 @@ public NonBlockingStatsDClientBuilder originDetectionEnabled(boolean val) { return this; } + public NonBlockingStatsDClientBuilder enableJdkSocket(boolean val) { + enableJdkSocket = val; + return this; + } + /** * Request that all metrics from this client to be enriched to specified tag cardinality. * @@ -455,7 +463,8 @@ private Callable getAddressLookupFromUrl(String url) { String uriPath = parsed.getPath(); return staticUnixResolution( uriPath, - UnixSocketAddressWithTransport.TransportType.fromScheme(parsed.getScheme())); + UnixSocketAddressWithTransport.TransportType.fromScheme(parsed.getScheme()), + enableJdkSocket); } return null; @@ -520,10 +529,24 @@ public SocketAddress call() { protected static Callable staticUnixResolution( final String path, final UnixSocketAddressWithTransport.TransportType transportType) { + return staticUnixResolution(path, transportType, false); + } + + private static Callable staticUnixResolution( + final String path, + final UnixSocketAddressWithTransport.TransportType transportType, + final boolean enableJdkSocket) { return new Callable() { @Override public SocketAddress call() { - final UnixSocketAddress socketAddress = new UnixSocketAddress(path); + SocketAddress socketAddress = + VersionUtils.isJavaVersionAtLeast(16) + && enableJdkSocket + && transportType + == UnixSocketAddressWithTransport.TransportType + .UDS_STREAM + ? VersionUtils.newUnixDomainSocketAddress(path) + : new UnixSocketAddress(path); return new UnixSocketAddressWithTransport(socketAddress, transportType); } }; diff --git a/src/main/java/com/timgroup/statsd/UnixDatagramClientChannel.java b/src/main/java/com/timgroup/statsd/UnixDatagramClientChannel.java index 4fccddf6..5b266ec0 100644 --- a/src/main/java/com/timgroup/statsd/UnixDatagramClientChannel.java +++ b/src/main/java/com/timgroup/statsd/UnixDatagramClientChannel.java @@ -16,6 +16,10 @@ class UnixDatagramClientChannel extends DatagramClientChannel { */ UnixDatagramClientChannel(SocketAddress address, int timeout, int bufferSize) throws IOException { + // Ideally we could use native JDK UDS support such as with the UnixStreamClientChannel. + // However, DatagramChannels do not support StandardProtocolFamily.UNIX, so this is + // unavailable. + // See this open issue for updates: https://bugs.openjdk.org/browse/JDK-8297837 super(UnixDatagramChannel.open(), address); // Set send timeout, to handle the case where the transmission buffer is full // If no timeout is set, the send becomes blocking diff --git a/src/main/java/com/timgroup/statsd/UnixStreamClientChannel.java b/src/main/java/com/timgroup/statsd/UnixStreamClientChannel.java index c86c3c57..86cd4ea2 100644 --- a/src/main/java/com/timgroup/statsd/UnixStreamClientChannel.java +++ b/src/main/java/com/timgroup/statsd/UnixStreamClientChannel.java @@ -2,8 +2,11 @@ import java.io.IOException; import java.net.SocketAddress; +import java.net.StandardSocketOptions; import java.nio.ByteBuffer; import java.nio.ByteOrder; +import java.nio.channels.SelectionKey; +import java.nio.channels.Selector; import java.nio.channels.SocketChannel; import jnr.unixsocket.UnixSocketAddress; import jnr.unixsocket.UnixSocketChannel; @@ -11,10 +14,11 @@ /** A ClientChannel for Unix domain sockets. */ public class UnixStreamClientChannel implements ClientChannel { - private final UnixSocketAddress address; + private final SocketAddress address; private final int timeout; private final int connectionTimeout; private final int bufferSize; + private final boolean enableJdkSocket; private SocketChannel delegate; private final ByteBuffer delimiterBuffer = @@ -26,13 +30,18 @@ public class UnixStreamClientChannel implements ClientChannel { * @param address Location of named pipe */ UnixStreamClientChannel( - SocketAddress address, int timeout, int connectionTimeout, int bufferSize) + SocketAddress address, + int timeout, + int connectionTimeout, + int bufferSize, + boolean enableJdkSocket) throws IOException { this.delegate = null; - this.address = (UnixSocketAddress) address; + this.address = address; this.timeout = timeout; this.connectionTimeout = connectionTimeout; this.bufferSize = bufferSize; + this.enableJdkSocket = enableJdkSocket; } @Override @@ -54,7 +63,7 @@ public synchronized int write(ByteBuffer src) throws IOException { delimiterBuffer.flip(); try { - long deadline = System.nanoTime() + timeout * 1_000_000L; + long deadline = timeout > 0 ? System.nanoTime() + timeout * 1_000_000L : 0; written = writeAll(delimiterBuffer, true, deadline); if (written > 0) { written += writeAll(src, false, deadline); @@ -87,19 +96,47 @@ public int writeAll(ByteBuffer bb, boolean canReturnOnTimeout, long deadline) throws IOException { int remaining = bb.remaining(); int written = 0; + long timeoutMs = timeout; + while (remaining > 0) { int read = delegate.write(bb); - - // If we haven't written anything yet, we can still return - if (read == 0 && canReturnOnTimeout && written == 0) { - return written; + if (read > 0) { + remaining -= read; + written += read; + if (deadline > 0 && System.nanoTime() >= deadline) { + throw new IOException("Write timed out"); + } + continue; } - remaining -= read; - written += read; + if (read == 0) { + if (delegate.isBlocking()) { + if (canReturnOnTimeout && written == 0) { + return written; + } + throw new IOException("Write timed out"); + } + + try (Selector selector = Selector.open()) { + delegate.register(selector, SelectionKey.OP_WRITE); + long selectTimeout = timeoutMs; - if (deadline > 0 && System.nanoTime() > deadline) { - throw new IOException("Write timed out"); + if (deadline > 0) { + long remainingNs = deadline - System.nanoTime(); + if (remainingNs <= 0) { + throw new IOException("Write timed out"); + } + long remainingMs = Math.max(1L, remainingNs / 1_000_000L); + selectTimeout = + timeoutMs > 0 ? Math.min(timeoutMs, remainingMs) : remainingMs; + } + + int selected = + selectTimeout > 0 ? selector.select(selectTimeout) : selector.select(); + if (selected == 0) { + throw new IOException("Write timed out after " + selectTimeout + "ms"); + } + } } } return written; @@ -127,40 +164,129 @@ private void connect() throws IOException { } } - UnixSocketChannel delegate = UnixSocketChannel.create(); - + // Use native JDK support for UDS on Java 16+ and jnr-unixsocket otherwise + if (VersionUtils.isJavaVersionAtLeast(16) && enableJdkSocket && connectWithJdkSocket()) { + return; + } + // Default to jnr-unixsocket if Java version is < 16 or native support is disabled + UnixSocketChannel channel = UnixSocketChannel.create(); long deadline = System.nanoTime() + connectionTimeout * 1_000_000L; + if (connectionTimeout > 0) { // Set connect timeout, this should work at least on linux // https://elixir.bootlin.com/linux/v5.7.4/source/net/unix/af_unix.c#L1696 - // We'd have better timeout support if we used Java 16's native Unix domain socket - // support (JEP 380) - delegate.setOption(UnixSocketOptions.SO_SNDTIMEO, connectionTimeout); + channel.setOption(UnixSocketOptions.SO_SNDTIMEO, connectionTimeout); } + try { - if (!delegate.connect(address)) { + UnixSocketAddress unixAddress; + if (address instanceof UnixSocketAddress) { + unixAddress = (UnixSocketAddress) address; + } else { + unixAddress = new UnixSocketAddress(address.toString()); + } + + if (!channel.connect(unixAddress)) { if (connectionTimeout > 0 && System.nanoTime() > deadline) { throw new IOException("Connection timed out"); } - if (!delegate.finishConnect()) { + if (!channel.finishConnect()) { throw new IOException("Connection failed"); } } - delegate.setOption(UnixSocketOptions.SO_SNDTIMEO, Math.max(timeout, 0)); + channel.setOption(UnixSocketOptions.SO_SNDTIMEO, Math.max(timeout, 0)); if (bufferSize > 0) { - delegate.setOption(UnixSocketOptions.SO_SNDBUF, bufferSize); + channel.setOption(UnixSocketOptions.SO_SNDBUF, bufferSize); } } catch (Exception e) { try { - delegate.close(); + channel.close(); } catch (IOException __) { // ignore } throw e; } - this.delegate = delegate; + this.delegate = channel; + } + + private boolean connectWithJdkSocket() throws IOException { + SocketChannel channel = null; + SocketAddress connectAddress; + + try { + // Only SocketChannel.open(ProtocolFamily) needs reflection; connect/finishConnect + // have existed since Java 1.4 and are called directly once we have the channel. + channel = openJdkSocketChannel(); + connectAddress = nativeSocketAddress(address); + if (bufferSize > 0) { + channel.setOption(StandardSocketOptions.SO_SNDBUF, bufferSize); + } + } catch (Exception | LinkageError e) { + closeQuietly(channel); + return false; + } + + try { + if (connectionTimeout <= 0) { + channel.configureBlocking(true); + channel.connect(connectAddress); + channel.configureBlocking(false); + } else { + channel.configureBlocking(false); + long deadline = System.nanoTime() + connectionTimeout * 1_000_000L; + if (!channel.connect(connectAddress)) { + try (Selector selector = Selector.open()) { + channel.register(selector, SelectionKey.OP_CONNECT); + while (!channel.finishConnect()) { + long remainingNs = deadline - System.nanoTime(); + if (remainingNs <= 0) { + throw new IOException( + "Connection timed out after " + connectionTimeout + "ms"); + } + long selectTimeout = Math.max(1L, remainingNs / 1_000_000L); + if (selector.select(selectTimeout) == 0 + && System.nanoTime() >= deadline) { + throw new IOException( + "Connection timed out after " + connectionTimeout + "ms"); + } + selector.selectedKeys().clear(); + } + } + } + } + } catch (IOException | RuntimeException e) { + closeQuietly(channel); + throw e; + } + + this.delegate = channel; + return true; + } + + SocketChannel openJdkSocketChannel() throws IOException { + return VersionUtils.openUnixSocketChannel(); + } + + private static SocketAddress nativeSocketAddress(SocketAddress address) { + if (address instanceof UnixSocketAddressWithTransport) { + address = ((UnixSocketAddressWithTransport) address).getAddress(); + } + if (address instanceof UnixSocketAddress) { + return VersionUtils.newUnixDomainSocketAddress(((UnixSocketAddress) address).path()); + } + return address; + } + + private static void closeQuietly(SocketChannel channel) { + if (channel != null) { + try { + channel.close(); + } catch (IOException ignored) { + // ignore + } + } } @Override diff --git a/src/main/java/com/timgroup/statsd/VersionUtils.java b/src/main/java/com/timgroup/statsd/VersionUtils.java new file mode 100644 index 00000000..c9de9502 --- /dev/null +++ b/src/main/java/com/timgroup/statsd/VersionUtils.java @@ -0,0 +1,143 @@ +package com.timgroup.statsd; + +import java.io.IOException; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.net.ProtocolFamily; +import java.net.SocketAddress; +import java.nio.channels.SocketChannel; +import java.util.ArrayList; +import java.util.List; + +// Logic copied from dd-trace-java Platform class. See: +// https://github.com/DataDog/dd-trace-java/blob/master/internal-api/src/main/java/datadog/trace/api/Platform.java +final class VersionUtils { + private static final Version JAVA_VERSION = + parseJavaVersion(System.getProperty("java.version")); + + private VersionUtils() {} + + static Version parseJavaVersion(String javaVersion) { + // Remove pre-release part, usually -ea + final int indexOfDash = javaVersion.indexOf('-'); + if (indexOfDash >= 0) { + javaVersion = javaVersion.substring(0, indexOfDash); + } + + int major = 0; + int minor = 0; + int update = 0; + + try { + List nums = splitDigits(javaVersion); + major = nums.get(0); + + // for java 1.6/1.7/1.8 + if (major == 1) { + major = nums.get(1); + minor = nums.size() > 2 ? nums.get(2) : 0; + update = nums.size() > 3 ? nums.get(3) : 0; + } else { + minor = nums.size() > 1 ? nums.get(1) : 0; + update = nums.size() > 2 ? nums.get(2) : 0; + } + } catch (NumberFormatException | IndexOutOfBoundsException e) { + // unable to parse version string - do nothing + } + return new Version(major, minor, update); + } + + private static List splitDigits(String str) { + List results = new ArrayList<>(); + + int len = str.length(); + + int value = 0; + for (int i = 0; i < len; i++) { + char ch = str.charAt(i); + if (ch >= '0' && ch <= '9') { + value = value * 10 + (ch - '0'); + } else if (ch == '.' || ch == '_' || ch == '+') { + results.add(value); + value = 0; + } else { + throw new NumberFormatException(); + } + } + results.add(value); + return results; + } + + static final class Version { + final int major; + final int minor; + final int update; + + Version(int major, int minor, int update) { + this.major = major; + this.minor = minor; + this.update = update; + } + + boolean isAtLeast(int major, int minor, int update) { + return isAtLeast(this.major, this.minor, this.update, major, minor, update); + } + + private static boolean isAtLeast( + int major, + int minor, + int update, + int atLeastMajor, + int atLeastMinor, + int atLeastUpdate) { + return (major > atLeastMajor) + || (major == atLeastMajor && minor > atLeastMinor) + || (major == atLeastMajor && minor == atLeastMinor && update >= atLeastUpdate); + } + } + + static boolean isJavaVersionAtLeast(int major) { + return JAVA_VERSION.isAtLeast(major, 0, 0); + } + + /** + * Opens a {@link SocketChannel} for Unix domain sockets using {@code + * StandardProtocolFamily.UNIX}, available since Java 16. Uses reflection to avoid a + * compile-time dependency on Java 16+ classes. + */ + @SuppressWarnings("unchecked") + static SocketChannel openUnixSocketChannel() throws IOException { + try { + Class standardProtocolFamilyClass = Class.forName("java.net.StandardProtocolFamily"); + Object unixProtocol = Enum.valueOf((Class) standardProtocolFamilyClass, "UNIX"); + Method openMethod = SocketChannel.class.getMethod("open", ProtocolFamily.class); + return (SocketChannel) openMethod.invoke(null, unixProtocol); + } catch (InvocationTargetException e) { + Throwable cause = e.getCause(); + if (cause instanceof IOException) { + throw (IOException) cause; + } + throw new IOException("Failed to open Unix domain SocketChannel", e); + } catch (Exception e) { + throw new IOException("Failed to open Unix domain SocketChannel", e); + } + } + + /** + * Creates a {@code java.net.UnixDomainSocketAddress} for the given path using reflection. + * Available since Java 16. + */ + static SocketAddress newUnixDomainSocketAddress(String path) { + try { + Class cls = Class.forName("java.net.UnixDomainSocketAddress"); + Method of = cls.getMethod("of", String.class); + return (SocketAddress) of.invoke(null, path); + } catch (InvocationTargetException e) { + throw new StatsDClientException( + "Failed to create UnixDomainSocketAddress for path: " + path, e.getCause()); + } catch (Exception e) { + throw new StatsDClientException( + "Failed to create UnixDomainSocketAddress for path: " + path, e); + } + } +} diff --git a/src/test/java/com/timgroup/statsd/BuilderAddressTest.java b/src/test/java/com/timgroup/statsd/BuilderAddressTest.java index d37703d5..afe0e136 100644 --- a/src/test/java/com/timgroup/statsd/BuilderAddressTest.java +++ b/src/test/java/com/timgroup/statsd/BuilderAddressTest.java @@ -1,6 +1,8 @@ package com.timgroup.statsd; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import java.net.InetSocketAddress; import java.net.SocketAddress; @@ -213,10 +215,20 @@ public void address_resolution() throws Exception { if (expected instanceof UnixSocketAddressWithTransport) { UnixSocketAddressWithTransport a = (UnixSocketAddressWithTransport) actual; UnixSocketAddressWithTransport e = (UnixSocketAddressWithTransport) expected; + // Native JDK UDS support returns a SocketAddress rather than a UnixSocketAddress assertEquals( ((FakeUnixSocketAddress) e.getAddress()).getPath(), - ((UnixSocketAddress) a.getAddress()).path()); + a.getAddress() instanceof UnixSocketAddress + ? ((UnixSocketAddress) a.getAddress()).path() + : a.getAddress().toString()); assertEquals(e.getTransportType(), a.getTransportType()); + if (VersionUtils.isJavaVersionAtLeast(16) + && e.getTransportType() + == UnixSocketAddressWithTransport.TransportType.UDS_STREAM) { + assertFalse(a.getAddress() instanceof UnixSocketAddress); + } else { + assertTrue(a.getAddress() instanceof UnixSocketAddress); + } } else { assertEquals(expected, actual); } diff --git a/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientTest.java b/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientTest.java index 8ed7a030..4895657d 100644 --- a/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientTest.java +++ b/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientTest.java @@ -1285,7 +1285,8 @@ ClientChannel createByteChannel( Callable addressLookup, int timeout, int connectionTimeout, - int bufferSize) + int bufferSize, + boolean enableJdkSocket) throws Exception { return new DatagramClientChannel(addressLookup.call()) { @Override @@ -1336,7 +1337,8 @@ ClientChannel createByteChannel( Callable addressLookup, int timeout, int connectionTimeout, - int bufferSize) + int bufferSize, + boolean enableJdkSocket) throws Exception { return new DatagramClientChannel(addressLookup.call()) { @Override diff --git a/src/test/java/com/timgroup/statsd/UnixStreamSocketDummyStatsDServer.java b/src/test/java/com/timgroup/statsd/UnixStreamSocketDummyStatsDServer.java index 6c480a6a..56a425a7 100644 --- a/src/test/java/com/timgroup/statsd/UnixStreamSocketDummyStatsDServer.java +++ b/src/test/java/com/timgroup/statsd/UnixStreamSocketDummyStatsDServer.java @@ -23,7 +23,8 @@ public class UnixStreamSocketDummyStatsDServer extends DummyStatsDServer { public UnixStreamSocketDummyStatsDServer(String socketPath) throws IOException { server = UnixServerSocketChannel.open(); server.configureBlocking(true); - server.socket().bind(new UnixSocketAddress(socketPath)); + UnixSocketAddress address = new UnixSocketAddress(socketPath); + server.socket().bind(address); this.listen(); } diff --git a/src/test/java/com/timgroup/statsd/UnixStreamSocketTest.java b/src/test/java/com/timgroup/statsd/UnixStreamSocketTest.java index d7e64e4d..3e193f41 100644 --- a/src/test/java/com/timgroup/statsd/UnixStreamSocketTest.java +++ b/src/test/java/com/timgroup/statsd/UnixStreamSocketTest.java @@ -10,6 +10,9 @@ import java.io.File; import java.io.IOException; +import java.net.SocketAddress; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; import java.nio.file.Files; import java.util.logging.Logger; import org.junit.After; @@ -95,11 +98,14 @@ public void assert_default_uds_size() throws Exception { @Test(timeout = 5000L) public void sends_to_statsd() throws Exception { + Thread.sleep(100); + server.clear(); + for (long i = 0; i < 5; i++) { client.gauge("mycount", i); server.waitForMessage(); String expected = String.format("my.prefix.mycount:%d|g", i); - assertThat(server.messagesReceived(), contains(expected)); + assertThat(server.messagesReceived(), hasItem(expected)); server.clear(); } assertThat(lastException.getMessage(), nullValue()); @@ -182,6 +188,38 @@ public void resist_dsd_timeout() throws Exception { server.clear(); } + @Test(timeout = 5000L) + public void nativeSocketAcceptsLegacyUnixAddress() throws Exception { + Assume.assumeTrue(VersionUtils.isJavaVersionAtLeast(16)); + UnixStreamClientChannel channel = + new UnixStreamClientChannel( + legacyUnixSocketAddress(socketFile.getPath()), 500, 500, -1, true); + + assertChannelSends(channel, "legacy.address:1|c"); + } + + @Test(timeout = 5000L) + public void fallsBackWhenNativeSocketCannotBeOpened() throws Exception { + Assume.assumeTrue(VersionUtils.isJavaVersionAtLeast(16)); + UnixStreamClientChannel channel = + new UnixStreamClientChannel( + legacyUnixSocketAddress(socketFile.getPath()), 500, 500, -1, true) { + @Override + SocketChannel openJdkSocketChannel() throws IOException { + throw new IOException("Native UDS unavailable"); + } + }; + + assertChannelSends(channel, "fallback:1|c"); + } + + private static SocketAddress legacyUnixSocketAddress(String path) throws Exception { + return (SocketAddress) + Class.forName("jnr.unixsocket.UnixSocketAddress") + .getConstructor(String.class) + .newInstance(path); + } + @Test(timeout = 5000L) public void stream_uds_has_uds_buffer_size() throws Exception { final NonBlockingStatsDClient client = @@ -208,4 +246,16 @@ public void max_packet_size_override() throws Exception { assertEquals(client.statsDProcessor.bufferPool.getBufferSize(), 576); } + + private static void assertChannelSends(UnixStreamClientChannel channel, String message) + throws Exception { + try { + channel.write(ByteBuffer.wrap(message.getBytes(NonBlockingStatsDClient.UTF_8))); + server.waitForMessage(); + assertThat(server.messagesReceived(), hasItem(message)); + server.clear(); + } finally { + channel.close(); + } + } } diff --git a/src/test/java/com/timgroup/statsd/VersionUtilsTest.java b/src/test/java/com/timgroup/statsd/VersionUtilsTest.java new file mode 100644 index 00000000..40fd3c85 --- /dev/null +++ b/src/test/java/com/timgroup/statsd/VersionUtilsTest.java @@ -0,0 +1,45 @@ +package com.timgroup.statsd; + +import static org.junit.Assert.assertEquals; + +import org.junit.Test; + +public class VersionUtilsTest { + @Test + public void parsesLegacyJavaVersions() { + assertVersion("1.8.0_382", 8, 0, 382); + assertVersion("1.8.0", 8, 0, 0); + assertVersion("1.7", 7, 0, 0); + } + + @Test + public void parsesModernJavaVersions() { + assertVersion("16", 16, 0, 0); + assertVersion("17.0", 17, 0, 0); + assertVersion("17.0.10+7", 17, 0, 10); + assertVersion("21-ea", 21, 0, 0); + } + + @Test + public void rejectsUnrecognizedJavaVersions() { + assertVersion("not-a-version", 0, 0, 0); + } + + @Test + public void comparesVersions() { + VersionUtils.Version version = VersionUtils.parseJavaVersion("17.0.10"); + + assertEquals(true, version.isAtLeast(17, 0, 10)); + assertEquals(true, version.isAtLeast(16, 0, 0)); + assertEquals(false, version.isAtLeast(17, 0, 11)); + } + + private static void assertVersion( + String input, int expectedMajor, int expectedMinor, int expectedUpdate) { + VersionUtils.Version version = VersionUtils.parseJavaVersion(input); + + assertEquals(expectedMajor, version.major); + assertEquals(expectedMinor, version.minor); + assertEquals(expectedUpdate, version.update); + } +}