Merge branch 'release/0.12.2-RC3'

Author: robertroeser

.travis.yml

@@ -16,9 +16,13 @@
 ---
 language: java
 
-jdk:
-- oraclejdk8
-# - oraclejdk9
+matrix:
+  include:
+    - jdk: oraclejdk8
+    - jdk: openjdk11
+      env: SKIP_RELEASE=true
+    - jdk: openjdk12
+      env: SKIP_RELEASE=true
 
 env:
   global:
@@ -37,4 +41,3 @@ cache:
   directories:
   - $HOME/.gradle/caches/
   - $HOME/.gradle/wrapper/
-

README.md

@@ -23,10 +23,10 @@ Example:
 
 ```groovy
 dependencies {
-    implementation 'io.rsocket:rsocket-core:0.11.14'
-    implementation 'io.rsocket:rsocket-transport-netty:0.11.14'
-//    implementation 'io.rsocket:rsocket-core:0.11.15.BUILD-SNAPSHOT'
-//    implementation 'io.rsocket:rsocket-transport-netty:0.11.15.BUILD-SNAPSHOT'
+    implementation 'io.rsocket:rsocket-core:0.12.2-RC2'
+    implementation 'io.rsocket:rsocket-transport-netty:0.12.2-RC2'
+//    implementation 'io.rsocket:rsocket-core:0.12.2-RC3-SNAPSHOT'
+//    implementation 'io.rsocket:rsocket-transport-netty:0.12.2-RC3-SNAPSHOT'
 }
 ```
 
@@ -91,7 +91,7 @@ or you will get a memory leak. Used correctly this will reduce latency and incre
 ```java
 RSocketFactory.receive()
         // Enable Zero Copy
-        .payloadDecoder(Frame::retain)
+        .frameDecoder(PayloadDecoder.ZERO_COPY)
         .acceptor(new PingHandler())
         .transport(TcpServerTransport.create(7878))
         .start()
@@ -105,7 +105,7 @@ RSocketFactory.receive()
 Mono<RSocket> client =
         RSocketFactory.connect()
             // Enable Zero Copy
-            .payloadDecoder(Frame::retain)
+            .frameDecoder(PayloadDecoder.ZERO_COPY)
             .transport(TcpClientTransport.create(7878))
             .start();
 ```

artifactory.gradle

@@ -35,6 +35,9 @@ if (project.hasProperty('bintrayUser') && project.hasProperty('bintrayKey')) {
                     }
                 }
             }
+            tasks.named("artifactoryPublish").configure {
+                onlyIf { System.getenv('SKIP_RELEASE') != "true" }
+            }
         }
     }
 }

bintray.gradle

@@ -55,6 +55,9 @@ if (project.hasProperty('bintrayUser') && project.hasProperty('bintrayKey') &&
                     }
                 }
             }
+            tasks.named("bintrayUpload").configure {
+                onlyIf { System.getenv('SKIP_RELEASE') != "true" }
+            }
         }
     }
 }

build.gradle

@@ -29,11 +29,11 @@ subprojects {
     apply plugin: 'io.spring.dependency-management'
     apply plugin: 'com.github.sherter.google-java-format'
 
-    ext['reactor-bom.version'] = 'Californium-SR5'
+    ext['reactor-bom.version'] = 'Californium-SR8'
     ext['logback.version'] = '1.2.3'
     ext['findbugs.version'] = '3.0.2'
-    ext['netty.version'] = '4.1.31.Final'
-    ext['netty-boringssl.version'] = '2.0.18.Final'
+    ext['netty.version'] = '4.1.36.Final'
+    ext['netty-boringssl.version'] = '2.0.25.Final'
     ext['hdrhistogram.version'] = '2.1.10'
     ext['mockito.version'] = '2.25.1'
     ext['slf4j.version'] = '1.7.25'
@@ -116,6 +116,10 @@ subprojects {
 
             systemProperty "io.netty.leakDetection.level", "ADVANCED"
         }
+        
+        tasks.named("javadoc").configure {
+            onlyIf { System.getenv('SKIP_RELEASE') != "true" }
+        }
     }
 
     plugins.withType(JavaLibraryPlugin) {

gradle.properties

@@ -11,4 +11,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-version=0.12.2-RC2
+version=0.12.2-RC3

rsocket-core/src/main/java/io/rsocket/RSocketClient.java

@@ -16,16 +16,23 @@
 
 package io.rsocket;
 
+import static io.rsocket.keepalive.KeepAliveSupport.ClientKeepAliveSupport;
+import static io.rsocket.keepalive.KeepAliveSupport.KeepAlive;
+
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.collection.IntObjectHashMap;
+import io.rsocket.exceptions.ConnectionErrorException;
 import io.rsocket.exceptions.Exceptions;
 import io.rsocket.frame.*;
 import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.internal.LimitableRequestPublisher;
 import io.rsocket.internal.UnboundedProcessor;
 import io.rsocket.internal.UnicastMonoProcessor;
+import io.rsocket.keepalive.KeepAliveFramesAcceptor;
+import io.rsocket.keepalive.KeepAliveHandler;
+import io.rsocket.keepalive.KeepAliveSupport;
 import java.nio.channels.ClosedChannelException;
 import java.util.Collections;
 import java.util.Map;
@@ -36,11 +43,7 @@
 import org.reactivestreams.Processor;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
-import reactor.core.publisher.BaseSubscriber;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.core.publisher.SignalType;
-import reactor.core.publisher.UnicastProcessor;
+import reactor.core.publisher.*;
 
 /** Client Side of a RSocket socket. Sends {@link ByteBuf}s to a {@link RSocketServer} */
 class RSocketClient implements RSocket {
@@ -54,14 +57,18 @@ class RSocketClient implements RSocket {
   private final UnboundedProcessor<ByteBuf> sendProcessor;
   private final Lifecycle lifecycle = new Lifecycle();
   private final ByteBufAllocator allocator;
+  private final KeepAliveFramesAcceptor keepAliveFramesAcceptor;
 
   /*client requester*/
   RSocketClient(
       ByteBufAllocator allocator,
       DuplexConnection connection,
       PayloadDecoder payloadDecoder,
       Consumer<Throwable> errorConsumer,
-      StreamIdSupplier streamIdSupplier) {
+      StreamIdSupplier streamIdSupplier,
+      int keepAliveTickPeriod,
+      int keepAliveAckTimeout,
+      KeepAliveHandler keepAliveHandler) {
     this.allocator = allocator;
     this.connection = connection;
     this.payloadDecoder = payloadDecoder;
@@ -74,19 +81,40 @@ class RSocketClient implements RSocket {
     this.sendProcessor = new UnboundedProcessor<>();
 
     connection.onClose().doFinally(signalType -> terminate()).subscribe(null, errorConsumer);
-
-    sendProcessor
-        .doOnRequest(
-            r -> {
-              for (LimitableRequestPublisher lrp : senders.values()) {
-                lrp.increaseInternalLimit(r);
-              }
-            })
-        .transform(connection::send)
+    connection
+        .send(sendProcessor)
         .doFinally(this::handleSendProcessorCancel)
         .subscribe(null, this::handleSendProcessorError);
 
     connection.receive().subscribe(this::handleIncomingFrames, errorConsumer);
+
+    if (keepAliveTickPeriod != 0 && keepAliveHandler != null) {
+      KeepAliveSupport keepAliveSupport =
+          new ClientKeepAliveSupport(allocator, keepAliveTickPeriod, keepAliveAckTimeout);
+      this.keepAliveFramesAcceptor =
+          keepAliveHandler.start(keepAliveSupport, sendProcessor::onNext, this::terminate);
+    } else {
+      keepAliveFramesAcceptor = null;
+    }
+  }
+
+  /*server requester*/
+  RSocketClient(
+      ByteBufAllocator allocator,
+      DuplexConnection connection,
+      PayloadDecoder payloadDecoder,
+      Consumer<Throwable> errorConsumer,
+      StreamIdSupplier streamIdSupplier) {
+    this(allocator, connection, payloadDecoder, errorConsumer, streamIdSupplier, 0, 0, null);
+  }
+
+  private void terminate(KeepAlive keepAlive) {
+    String message =
+        String.format("No keep-alive acks for %d ms", keepAlive.getTimeout().toMillis());
+    ConnectionErrorException err = new ConnectionErrorException(message);
+    lifecycle.setTerminationError(err);
+    errorConsumer.accept(err);
+    connection.dispose();
   }
 
   private void handleSendProcessorError(Throwable t) {
@@ -294,7 +322,7 @@ public void accept(long n) {
                             .transform(
                                 f -> {
                                   LimitableRequestPublisher<Payload> wrapped =
-                                      LimitableRequestPublisher.wrap(f, sendProcessor.available());
+                                      LimitableRequestPublisher.wrap(f);
                                   // Need to set this to one for first the frame
                                   wrapped.request(1);
                                   senders.put(streamId, wrapped);
@@ -452,8 +480,9 @@ private void handleStreamZero(FrameType type, ByteBuf frame) {
       case LEASE:
         break;
       case KEEPALIVE:
-        // KeepAlive is handled by corresponding connection interceptor,
-        // just release its frame here
+        if (keepAliveFramesAcceptor != null) {
+          keepAliveFramesAcceptor.receive(frame);
+        }
         break;
       default:
         // Ignore unknown frames. Throwing an error will close the socket.