Merge branch 'release/0.8.9'

Author: rdegnan

.travis.yml

@@ -33,7 +33,7 @@ cache:
   - "$HOME/.m2"
 env:
   global:
-  - secure: fG6x4/LuQuYDO6B3hw9YnxTKBepOS/zcSqDbyl3fJkYSDYuRS6yvI0CsQCQGTLtD3MKED+0o43rUiuPgNzz3qENKQukrFgmXQl24fwuUxtFhz4Lbb3OCMnWMAU3SgWZmpbUoOhCEGr0LzvYnb+MImiRU2v7gxr4Nc9hy8gRZjQQ4R7V+UlhOhZVjN5JfsEnKMpZejEkFlNADCdAhSDdHLVZN7lOsFjDC7XQ/GKLKGrRyIzREto9hm58QIsolghMyIEL2Jj1+ePTM/3wCYLUaRpdUjmHQawJi9vfwFO4g4yyKdUVXHpITQwQ7JE/SoJsje8ZUhOJZ3TvaBDlVjvXP41P3JNDbSesrBYcR0k70TYr7wliHKsdGurHvDQ/2Rxas/FITFGvLchYczbUPLJ7h0fJha+cpwsmmNvLmg0RIaaHa4RXPknWuetipbJFSqYtBN01iCtdSzHwwhZGZ6zV3kPBa0QWsDLWa/Ih2STDY6DVIA6YgaqaM6aN9U/UR8XeCnjPot+StaNuU2nSgGcwX5BIZRMuca9OF6FryG+t527njU86mp2QSj1HOTs/c31v9sSg2krCl7jChm3ZtpXyiEp+IdKaa5uZHLyx0fcm489375PLjhE6gRp/zvYf58uSvoMLOl1xwfOkNiPP8YXIDtVodzgGxCnxD49WE5B9CCbk=
-  - secure: NvHtXfxiqBwkmj5l7Wr9EuBi9X0dMVL5CCeKFVqzIOmXCIdjwvknQ3rfZZ46fjwOQC/FuCcoM/8dHkBIkGzqIVo9Qx1+AGnq4AO7rr+iLYPvmz2wB8QKrD81bVAhtoKnuYe1wyu1XvURBgrietjoAgxrhgW3iReBVW4dj0YaTdRZXAD70M3i63tfV9GyB2pv6NfRyaq50IGkaJPdiyUPpDWcSmtnrkxQvUvIzNm50AGKEq3IuLp+wgVouAI81m1a+B0TuEfo4VnqOKlJx1dq9VT76GN4Ni+u2Rnnli2kFAXGtGfuPrZF1WPE2APJQ4W9H7YeGzya38JCUkkwD2Buxoe7CsW7ZEVnGbFE4xsuc+vejvDkeGF4LLABzuGwcDxG70MPHoXpyd1ayY3Iek1zPn4/3nFTO40YreeaHDT3MG3BSZx6WkMfmqfnFsKJeX9g7zzZLGmOuniwGLOIwFpePtyhn/wQvfPG+UalnSk+pXyGWuhnurcgDlk7ASlHj4TZnDiZvCCZy1xdzytttwf/i71Ts+6Db+TZfggnymeYFRo19UpLTr15Rdkwk0CNrJlvruKkGkR5G2els14M37PhHaeIdGol3ZqARs9+7o9gnRW2cPaQ26IRgNSGQwRhGaJcAyoLNZejc97OHEga9r6B8cIJK6NGD+4m8GA7Sh4O3EM=
   - secure: crwXv5Q/iCtIIC81QWgWYNfuGJxSQQjVn83vvRoRMXmfSZ02NnQfDos+p1GxxbIXLUQ8NsaLc8cg0o2PQgsK/l0d7nDExix5Rt+AwzU4UOrUcZh46MgEeZDow4q6ySToOHaeKsaRnbTIpAwU28o9fnZ9ViaQ7s2uZWQZXYigGf0PN1ca9E8MJ6VEMvp34e2wJok6woaWG9ALCL2b5viofTv0zdHXWfGV4rdz9Rq0fThmFn4BlmHdcTwGt2ZLdWcEbzv0qNYNAiVvNJd4uWHOnjRceslAZ/JZxgIzSHK0ctNiZdc9rRNo8BHGQqKvmZqmbzUAU07uS0+iLBnm89KKJyQCXKulDv7NM0ShiFexMQ17TUOdlJtH65ypzUIJ+sbSOj2M+NhThUD8OQlebvjKF6t8M9f1NIXkkH/Xw2XJ146ANBcGyjALVFDIqlKIyCtIoY3/teQWccfbAzBglC24QBizpfgmdBEGOJXs2qR0NxzcIRNc1o9XPyT9BJxa+9JnHyHbF+LG07j1LfaaJSQBIYL7xjDBwEEIOEmP3UrISwjvs1fOQA3oYINguXtl7qi3wNylv35e5xw5acijcTTYHRSY6A5rPdokbK0fCQHCrHldBYDc8Bvy+GnjnYQHynjdWGbTnkvMgvYe2aHD+mH5pO5V7DRYHhnJsuMuscRIELc=
   - secure: ZMu9rhzTT1F6pyqqrd0TGhAeUGigAvGZT5rJv2ifZJEB8+FOQ+/ylJ/Q3Gq9Wx8quWVR/ocTWF8yEuA1BlZF1Z+tfOcCXOaYPrgMFNhgmETK9X1AQN1gahkTzLNWEZUzKwte5IwaDYDwMe54VcJpY9/f7R6ZyjKVEJOLCjnJ+YLR2rjttsug9ZurH3iLzdgRGMHyI55+dz83MkdumkkATFAeLnQepdL2n4ijD0NvuzvlvaU+Knz9AZCX2kSNDM+hn+K0fFDM4CBFduBi8gEYmRlnSXslHtfwuZRViZQaS4voRMAFmWPNCH67ya3f1voi41BWXgaGTW5He8XDjQtFcIphvykmMOsS4ka9u4K0LEbReoSBRwBUAhigGMFXB1lMup66+WvjvSvwS9rMVVGaXpVGjgpM5shWpiIEoWICWU+Pu5ihk9BWv61/tYNNw6u3QlXt1ZfjM50F7eoJ1ctfYqClrlD3iwrBJ47aIu4kH1HdCp9uxkxq+Cz75n2UdV6Py7eEW1N141fnZ/Iax8nzbNu4SxccsJn1yUv22LJCteW3//GqvRVa7hets587PFpfkyaraBLNBvh90/wouXK/yvYBZN4WK7VwnuCUmxTjTglT9mQGUgPOYNMDU/m7rDp2zDc+8aMEfQjWGktH93M0g1l9yhidc8AMQbyAaxkk73M=
+  - secure: hxmPq8HtqUN04I639LiDA6euoWW8ziWs70USv7nwRLSE0gV/oDrwFrDX5QJNP4rP04hD/UCll0jKZPHClVjXfW7s9U2wqrFlyJifFoefQms7HZIselUuRXAjkK7KVA8ur1kz+6Ox3Ji7gxUnkF9cGR+rGAHyllSmuD+UIw2vCUHcqAAlhYzN9NfenjqEBNTQ5QTR0p1I37fDLsxORxTjuUxQO0yaYv1sy0eUUbDSR0XN0VrN/sNFMkV6+m5+A+7/4LYqxmZxyRqSC4Qk726k8Eab0K/QCnxE3TO/QoAPk9QZ2BEtHbrycNhm9yxG9+CGNbZCpQaWH1BEeo5F9B6u1prbikqzd9cTSuLGqR1r4wBDuIDFB8RloOMvoB0Z5+CWeP8X9u7q13vFFDS26crMWbqCCoo9Vcel8Js1H8/awnYrxtYhBk4BRGDzgUfRzaru+GpjE3s1kppkjj01qUo3fN9P+Px/zISPD9OdMgVvdN0O3rGPYHniFS9lQJZaJSJUU8hlApqsTQfK+5j8vIKmPe0ur39XriVeC8Kam8SWMymjLrLaNGOe0XvST9MLKKel7Aj9mknNyZ0dQEr24BLlN/vGHQsHTrpnReLHP3szuF5FXZQsqYQ2uN4Q5000O6IJfbD9FcBJHzOCtyIlNkJ12qK6GiLChwqcpv9muQZcaH8=
+  - secure: fyh1tUQJBijuGd5jO6ZHl2yn8u5uKgz+9tKgrGNBg2NNlxXvq1TJqBj1/fkZonH3rxPAH4AnmxsR4ktWKhmTci16D3ATflpxVOSQYHdMqbsxpNt4e+um/lnFLp6WORLkcnFAsA0rUrH3eRJXPfZ6YgcY3L4tddiT0sqLZ1ECo5gA915CPN8ANioHzooWJgVBcbr34z2HcJudgKws4C+MoiTKv3C9Ijamg+30D/wgiyRJXPD/azb2KoVx+Bk5yIQuBOl3G/a/LMpK+kv/2jZQjUT/gnKLEvRu9H5xJIL41mFTRgmZxD9E0Rb0VeAaYICfo+4K2g2smXeA/Eb9lmwMQYiEKdG2IH2chjPggBM0hsB42pGc+mFDKaNCtQSVkbQkrg2CmDMwvBaL7IBFNUmlSbZfG1PeQluSNs9uECR0MFM/zyNECLGkcrf3kZXHMuQ+1jYPWf37CuiNbEnN8sQm65t9wqX3TqRB8rksGaThVvYizIon7zuuIrKINynGDEc3ygVUOeOBLT+U9AaF1jatkxuwiZ/VJaReYuJvH76wCUb6k91RQA9udK4n8h8aMomOmTyIWRSnEFObDs1RW6Lixh5PRShTKG8ffoY0vM5IO5i17dPFOmIRo99yqKiWPcfQLK8IP0SiCwc6w0pqAns6H/r8qBgk6BoXGnBT5kyzKr8=

client/src/main/java/io/netifi/proteus/Proteus.java

@@ -141,7 +141,7 @@ public static class Builder {
     private int poolSize = Runtime.getRuntime().availableProcessors();
     private Supplier<Tracer> tracerSupplier = () -> null;
 
-    private Builder clientTransportFactory(
+    public Builder clientTransportFactory(
         Function<SocketAddress, ClientTransport> clientTransportFactory) {
       this.clientTransportFactory = clientTransportFactory;
       return this;

gradle.properties

@@ -1,2 +1,2 @@
 group=io.netifi.proteus
-version=0.8.8
+version=0.8.9

gradle/java.gradle

@@ -97,7 +97,7 @@ test {
     }
 }
 
-if ("$version".contains('SNAPSHOT')) {
+if ("$version".contains('SNAPSHOT') || "$version".contains('RC')) {
     publishing {
         publications {
             mavenJava(MavenPublication) {
@@ -129,7 +129,7 @@ if ("$version".contains('SNAPSHOT')) {
 
     bintray {
         user = project.hasProperty('bintrayUser') ? project.property('bintrayUser') : System.getenv('BINTRAY_USER')
-        key = project.hasProperty('bintrayKey') ? project.property('bintrayKey') : System.getenv('f')
+        key = project.hasProperty('bintrayKey') ? project.property('bintrayKey') : System.getenv('BINTRAY_KEY')
 
         publications = ['mavenJava']
 

protobuf-rpc/build.gradle

@@ -231,7 +231,7 @@ artifacts {
 //}
 
 
-if ("$version".contains('SNAPSHOT')) {
+if ("$version".contains('SNAPSHOT') || "$version".contains('RC')) {
     publishing {
         publications {
             maven(MavenPublication) {

tracing-openzipkin/src/main/java/io/netifi/proteus/tracing/ProteusReporter.java

@@ -6,7 +6,7 @@
 import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxProcessor;
-import reactor.retry.Retry;
+import reactor.core.publisher.Mono;
 import zipkin2.Annotation;
 import zipkin2.Component;
 import zipkin2.Endpoint;
@@ -15,6 +15,8 @@
 
 import java.io.IOException;
 import java.time.Duration;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 class ProteusReporter extends Component implements Reporter<Span> {
   private static final Logger logger = LoggerFactory.getLogger(ProteusReporter.class);
@@ -27,9 +29,20 @@ class ProteusReporter extends Component implements Reporter<Span> {
     this.sink = DirectProcessor.<Span>create().serialize();
     this.group = group;
     this.destination = destination;
+    AtomicInteger count = new AtomicInteger();
+    AtomicLong lastRetry = new AtomicLong(System.currentTimeMillis());
     this.disposable =
         Flux.defer(() -> service.streamSpans(sink.onBackpressureLatest().map(this::mapSpan)))
-            .doOnError(throwable -> logger.error("error sending tracing data", throwable))
+            .onErrorResume(
+                throwable -> {
+                  if (System.currentTimeMillis() - lastRetry.getAndSet(System.currentTimeMillis()) > 30_000) {
+                    count.set(0);
+                  }
+                  
+                  int i = Math.min(30, count.incrementAndGet());
+                  logger.error("error sending tracing data", throwable);
+                  return Mono.delay(Duration.ofSeconds(i)).then(Mono.error(throwable));
+                })
             .retry()
             .subscribe();
   }

tracing-openzipkin/src/main/java/io/netifi/proteus/tracing/ProteusZipkinHttpBridge.java

@@ -28,26 +28,12 @@ public class ProteusZipkinHttpBridge implements ProteusTracingService {
   private final int port;
 
   private final String zipkinUrl;
-
   private HttpClient httpClient;
 
   public ProteusZipkinHttpBridge(String host, int port, String zipkinUrl) {
     this.zipkinUrl = zipkinUrl;
     this.host = host;
     this.port = port;
-    this.httpClient =
-        HttpClient.builder()
-            .options(
-                builder ->
-                    builder
-                        .compression(true)
-                        .poolResources(PoolResources.fixed("proteusZipkinBridge"))
-                        .option(ChannelOption.SO_KEEPALIVE, true)
-                        .option(ChannelOption.SO_TIMEOUT, 60_000)
-                        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30_000)
-                        .host(host)
-                        .port(port))
-            .build();
   }
 
   public ProteusZipkinHttpBridge(String host, int port) {
@@ -94,6 +80,28 @@ public static void main(String... args) {
     proteus.onClose().block();
   }
 
+  private synchronized HttpClient getClient() {
+    if (httpClient == null) {
+      this.httpClient =
+          HttpClient.builder()
+              .options(
+                  builder ->
+                      builder
+                          .compression(true)
+                          .poolResources(PoolResources.fixed("proteusZipkinBridge"))
+                          .option(ChannelOption.SO_KEEPALIVE, true)
+                          .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30_000)
+                          .host(host)
+                          .port(port))
+              .build();
+    }
+    return httpClient;
+  }
+
+  private synchronized void resetHttpClient() {
+    this.httpClient = null;
+  }
+
   @Override
   public Mono<Ack> streamSpans(Publisher<Span> messages, ByteBuf metadata) {
     return Flux.from(messages)
@@ -119,22 +127,19 @@ public Mono<Ack> streamSpans(Publisher<Span> messages, ByteBuf metadata) {
                .flatMap(stringMono -> stringMono)
         .concatMap(
             spans ->
-                httpClient
+                getClient()
                     .post(
                         zipkinUrl,
                         request -> {
                           request.addHeader("Content-Type", "application/json");
                           return request.sendString(Mono.just(spans));
                         })
-                    .doOnError(
-                        throwable ->
-                            logger.error(
-                                "error sending data to tracing data to url "
-                                    + zipkinUrl
-                                    + " and payload [\n" + spans +"\n]",
-                                throwable))
-                    .timeout(Duration.ofSeconds(30)),
-            256)
+                    .timeout(Duration.ofSeconds(30))
+                    .doOnError(throwable -> resetHttpClient()),
+            8)
+        .doOnError(
+            throwable ->
+                logger.error("error sending data to tracing data to url " + zipkinUrl, throwable))
         .then(Mono.never());
   }
 }

tracing-openzipkin/src/main/resources/log4j2.xml

@@ -18,7 +18,7 @@
         <logger name="io.rsocket.FrameLogger" level="ERROR" additivity="false">
             <AppenderRef ref="STDOUT"/>
         </logger>
-        <logger name="io.netifi.proteus.tracing" level="ERROR" additivity="false">
+        <logger name="io.netifi.proteus.tracing" level="INFO" additivity="false">
             <AppenderRef ref="STDOUT"/>
         </logger>
     </Loggers>