Merge pull request #24 from netifi-proteus/hotfix/1.5.2

Hotfix/1.5.2

Author: kbahr

gradle.properties

@@ -1,2 +1,2 @@
 group=io.netifi.proteus
-version=1.5.1
+version=1.5.2

proteus-client/src/main/java/io/netifi/proteus/DefaultProteusBrokerService.java

@@ -16,6 +16,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.Unpooled;
+import io.netty.util.ReferenceCountUtil;
 import io.opentracing.Tracer;
 import io.rsocket.Payload;
 import io.rsocket.RSocket;
@@ -24,6 +25,7 @@
 import io.rsocket.rpc.stats.Quantile;
 import io.rsocket.transport.ClientTransport;
 import io.rsocket.util.ByteBufPayload;
+import io.rsocket.util.DefaultPayload;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.time.Duration;
@@ -264,10 +266,25 @@ WeightedReconnectingRSocket createWeightedReconnectingRSocket() {
   }
 
   Payload getSetupPayload(String computedFromDestination) {
-    ByteBuf metadata =
-        DestinationSetupFlyweight.encode(
-            ByteBufAllocator.DEFAULT, computedFromDestination, group, accessKey, accessToken);
-    return ByteBufPayload.create(Unpooled.EMPTY_BUFFER, metadata);
+    return getSetupPayload(
+        ByteBufAllocator.DEFAULT, computedFromDestination, group, accessKey, accessToken);
+  }
+
+  static Payload getSetupPayload(
+      ByteBufAllocator alloc,
+      String computedFromDestination,
+      String group,
+      long accessKey,
+      ByteBuf accessToken) {
+    ByteBuf metadata = null;
+    try {
+      metadata =
+          DestinationSetupFlyweight.encode(
+              alloc, computedFromDestination, group, accessKey, accessToken);
+      return DefaultPayload.create(Unpooled.EMPTY_BUFFER, metadata);
+    } finally {
+      ReferenceCountUtil.safeRelease(metadata);
+    }
   }
 
   private ProteusSocket unwrappedDestination(String destination, String group) {

proteus-client/src/test/java/io/netifi/proteus/ProteusBrokerServiceTest.java

@@ -0,0 +1,68 @@
+package io.netifi.proteus;
+
+import io.netifi.proteus.frames.DestinationSetupFlyweight;
+import io.netty.buffer.*;
+import io.rsocket.Payload;
+import java.nio.charset.Charset;
+import java.util.Collection;
+import java.util.function.Function;
+import org.junit.*;
+
+public class ProteusBrokerServiceTest {
+
+  @Test
+  public void setupPayloadLeakTest() {
+    ByteBuf token = Unpooled.wrappedBuffer("token".getBytes(Charset.defaultCharset()));
+    PooledByteBufAllocator alloc = nonCachingAllocator();
+
+    for (int i = 0; i < 100000; i++) {
+      Payload payload =
+          DefaultProteusBrokerService.getSetupPayload(alloc, "foo", "bar", 123L, token);
+    }
+    Assert.assertEquals(0, directBuffersCount(alloc));
+    Assert.assertEquals(0, heapBuffersCount(alloc));
+  }
+
+  @Test
+  public void setupDecodeTest() {
+    String expectedToken = "token";
+    String expectedDest = "foo";
+    String expectedGroup = "bar";
+    long expectedKey = 123L;
+
+    ByteBuf token = Unpooled.wrappedBuffer(expectedToken.getBytes(Charset.defaultCharset()));
+
+    Payload payload =
+        DefaultProteusBrokerService.getSetupPayload(
+            ByteBufAllocator.DEFAULT, expectedDest, expectedGroup, expectedKey, token);
+    ByteBuf metadata = payload.sliceMetadata();
+    String actualDest = DestinationSetupFlyweight.destination(metadata);
+    String actualGroup = DestinationSetupFlyweight.group(metadata);
+    long actualAccessKey = DestinationSetupFlyweight.accessKey(metadata);
+    String actualAccessToken =
+        DestinationSetupFlyweight.accessToken(metadata).toString(Charset.defaultCharset());
+
+    Assert.assertEquals(expectedToken, actualAccessToken);
+    Assert.assertEquals(expectedDest, actualDest);
+    Assert.assertEquals(expectedGroup, actualGroup);
+    Assert.assertEquals(expectedKey, actualAccessKey);
+  }
+
+  private static PooledByteBufAllocator nonCachingAllocator() {
+    return new PooledByteBufAllocator(true, 1, 1, 8192, 11, 0, 0, 0);
+  }
+
+  private static long directBuffersCount(PooledByteBufAllocator alloc) {
+    return count(alloc, PooledByteBufAllocator::directArenas);
+  }
+
+  private static long heapBuffersCount(PooledByteBufAllocator alloc) {
+    return count(alloc, PooledByteBufAllocator::heapArenas);
+  }
+
+  private static long count(
+      PooledByteBufAllocator alloc,
+      Function<PooledByteBufAllocator, Collection<PoolArenaMetric>> f) {
+    return f.apply(alloc).stream().mapToLong(PoolArenaMetric::numActiveAllocations).sum();
+  }
+}