Merge branch 'release/1.6.0'

Author: kbahr

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

@@ -32,6 +32,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;

proteus-client/src/main/java/io/netifi/proteus/rsocket/WeightedReconnectingRSocket.java

@@ -198,11 +198,13 @@ private ClientRSocketFactory getClientFactory() {
     if (keepalive) {
       connect =
           connect
+              .keepAlive()
               .keepAliveTickPeriod(Duration.ofSeconds(tickPeriodSeconds))
               .keepAliveAckTimeout(Duration.ofSeconds(ackTimeoutSeconds))
               .keepAliveMissedAcks(missedAcks);
     } else {
       connect
+          .keepAlive()
           .keepAliveAckTimeout(Duration.ofSeconds(0))
           .keepAliveAckTimeout(Duration.ofSeconds(0))
           .keepAliveMissedAcks(missedAcks);
@@ -276,8 +278,7 @@ void connect() {
                                     })
                                 .subscribe();
                             setRSocket(rSocket);
-                          })
-                      .doFinally(signalType -> setupPayload.release());
+                          });
                 }))
         .doOnError(t -> logger.error("error trying to broker", t))
         .retry()