Merge remote-tracking branch 'origin/develop' into develop

Author: kbahr

artifactory.gradle

@@ -31,7 +31,7 @@ if (project.hasProperty('netifiArtifactoryUsername') && project.hasProperty('net
                     }
 
                     defaults {
-                        publications('maven')
+                        publications('mavenJava')
                     }
                 }
             }

build.gradle

@@ -1,21 +1,31 @@
+buildscript {
+    repositories {
+        maven {
+            url "https://plugins.gradle.org/m2/"
+        }
+    }
+    dependencies {
+        classpath "com.jfrog.bintray.gradle:gradle-bintray-plugin:1.8.4"
+        classpath "org.jfrog.buildinfo:build-info-extractor-gradle:4.8.1"
+    }
+}
+
 plugins {
     id 'com.gradle.build-scan' version '1.16' // declare before any other plugin
     id 'com.google.osdetector' version '1.4.0'
     id 'com.github.sherter.google-java-format' version '0.7.1' apply false
     id 'com.github.johnrengelman.shadow' version '2.0.1' apply false
     id 'me.champeau.gradle.jmh' version '0.4.7' apply false
     id 'io.morethan.jmhreport' version '0.6.2.1' apply false
-    id 'com.jfrog.artifactory' version '4.7.3'
-    id 'com.jfrog.bintray' version '1.8.4'
     id 'io.spring.dependency-management' version '1.0.6.RELEASE' apply false
     id 'com.google.protobuf' version '0.8.5' apply false
 }
 
-apply from: 'artifactory.gradle'
-apply from: 'bintray.gradle'
-
 buildScan { licenseAgreementUrl = 'https://gradle.com/terms-of-service'; licenseAgree = 'yes' }
 
+apply from: "artifactory.gradle"
+apply from: "bintray.gradle"
+
 subprojects {
     apply plugin: 'io.spring.dependency-management'
     apply plugin: 'com.jfrog.bintray'
@@ -24,7 +34,7 @@ subprojects {
     apply plugin: 'com.github.sherter.google-java-format'
     apply plugin: 'maven'
     apply plugin: 'maven-publish'
-
+    
     ext['reactor-bom.version'] = 'Californium-SR1'
     ext['rsocket.version'] = '0.11.9'
     ext['rsocket-rpc.version'] = '0.2.4'
@@ -45,7 +55,6 @@ subprojects {
     ext['zipkin.sender.version'] = '2.7.6'
     ext['hdrhistogram.version'] = '2.1.10'
     ext['netty.version'] = '4.1.29.Final'
-    ext['netty-tcnative.version'] = '2.0.18.Final'
     ext['typesafe-config.version'] = '1.3.3'
 
     sourceCompatibility = 1.8
@@ -77,10 +86,6 @@ subprojects {
                 entry 'protobuf-java-util'
             }
 
-            dependencySet( group: 'io.netty', version: ext['netty-tcnative.version']) {
-                entry 'netty-tcnative'
-            }
-
             dependency "com.hubspot.jackson:jackson-datatype-protobuf:${ext['jackson-protobuf.version']}"
             dependency "com.fasterxml.jackson.core:jackson-databind:${ext['jackson.version']}"
             dependency "com.google.guava:guava:${ext['guava.version']}"
@@ -163,4 +168,35 @@ subprojects {
         testCompile 'io.rsocket:rsocket-transport-local'
         testCompile 'org.mockito:mockito-core'
     }
+
+    // Build Source Jars
+    def sourcesJarMap = [name: 'sourcesJar', type: Jar, dependsOn: classes]
+    project.tasks.create(sourcesJarMap, {
+        classifier = 'sources'
+        from sourceSets.main.allSource
+        version = "${project.version}"
+    })
+
+    // Build Javadoc Jars
+    def javadocJarMap = [name: 'javadocJar', type: Jar, dependsOn: javadoc]
+    project.tasks.create(javadocJarMap, {
+        classifier = 'javadoc'
+        from javadoc.destinationDir
+        version = "${project.version}"
+    })
+
+    artifacts {
+        archives sourcesJar
+        archives javadocJar
+    }
+    
+    publishing {
+        publications {
+            mavenJava(MavenPublication) {
+                from components.java
+                artifact sourcesJar
+                artifact javadocJar
+            }
+        }
+    }
 }

proteus-client/build.gradle

@@ -10,7 +10,7 @@ dependencies {
     compile 'com.typesafe:config'
     compile 'com.google.guava:guava'
     compile 'io.rsocket.rpc:rsocket-rpc-core'
-    compile 'io.netty:netty-tcnative::linux-x86_64'
+    compile "io.netty:netty-tcnative:2.0.18.Final:linux-x86_64"
 
     testProtobuf 'io.rsocket.rpc:rsocket-rpc-protobuf-idl'
     testCompile project(':proteus-tracing-openzipkin')

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();
+  }
+}