Merge branch 'release/1.0.0-RC2'

Author: robertroeser

.travis.yml

@@ -16,6 +16,8 @@
 ---
 language: java
 
+dist: trusty
+
 matrix:
   include:
     - jdk: oraclejdk8

build.gradle

@@ -29,10 +29,10 @@ subprojects {
     apply plugin: 'io.spring.dependency-management'
     apply plugin: 'com.github.sherter.google-java-format'
 
-    ext['reactor-bom.version'] = 'Californium-SR8'
+    ext['reactor-bom.version'] = 'Dysprosium-M3'
     ext['logback.version'] = '1.2.3'
     ext['findbugs.version'] = '3.0.2'
-    ext['netty.version'] = '4.1.36.Final'
+    ext['netty.version'] = '4.1.37.Final'
     ext['netty-boringssl.version'] = '2.0.25.Final'
     ext['hdrhistogram.version'] = '2.1.10'
     ext['mockito.version'] = '2.25.1'
@@ -89,6 +89,7 @@ subprojects {
 
     repositories {
         mavenCentral()
+        maven { url 'http://repo.spring.io/milestone' } // temporary for Reactor Dysprosium
 
         if (version.endsWith('BUILD-SNAPSHOT') || project.hasProperty('platformVersion')) {
             maven { url 'http://repo.spring.io/libs-snapshot' }

gradle.properties

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

rsocket-core/src/jmh/java/io/rsocket/StreamIdSupplierPerf.java

@@ -0,0 +1,35 @@
+package io.rsocket;
+
+import io.netty.util.collection.IntObjectMap;
+import io.rsocket.internal.SynchronizedIntObjectHashMap;
+import org.openjdk.jmh.annotations.*;
+import org.openjdk.jmh.infra.Blackhole;
+
+@BenchmarkMode(Mode.Throughput)
+@Fork(
+    value = 1 // , jvmArgsAppend = {"-Dio.netty.leakDetection.level=advanced"}
+    )
+@Warmup(iterations = 10)
+@Measurement(iterations = 10)
+@State(Scope.Thread)
+public class StreamIdSupplierPerf {
+  @Benchmark
+  public void benchmarkStreamId(Input input) {
+    int i = input.supplier.nextStreamId(input.map);
+    input.bh.consume(i);
+  }
+
+  @State(Scope.Benchmark)
+  public static class Input {
+    Blackhole bh;
+    IntObjectMap map;
+    StreamIdSupplier supplier;
+
+    @Setup
+    public void setup(Blackhole bh) {
+      this.supplier = StreamIdSupplier.clientSupplier();
+      this.bh = bh;
+      this.map = new SynchronizedIntObjectHashMap();
+    }
+  }
+}

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

@@ -20,6 +20,7 @@
 import io.netty.util.AbstractReferenceCounted;
 import io.rsocket.frame.FrameHeaderFlyweight;
 import io.rsocket.frame.SetupFrameFlyweight;
+import javax.annotation.Nullable;
 
 /**
  * Exposed to server for determination of ResponderRSocket based on mime types and SETUP
@@ -43,6 +44,11 @@ public static ConnectionSetupPayload create(final ByteBuf setupFrame) {
 
   public abstract boolean willClientHonorLease();
 
+  public abstract boolean isResumeEnabled();
+
+  @Nullable
+  public abstract ByteBuf resumeToken();
+
   @Override
   public ConnectionSetupPayload retain() {
     super.retain();
@@ -101,6 +107,16 @@ public boolean willClientHonorLease() {
       return SetupFrameFlyweight.honorLease(setupFrame);
     }
 
+    @Override
+    public boolean isResumeEnabled() {
+      return SetupFrameFlyweight.resumeEnabled(setupFrame);
+    }
+
+    @Override
+    public ByteBuf resumeToken() {
+      return SetupFrameFlyweight.resumeToken(setupFrame);
+    }
+
     @Override
     public ConnectionSetupPayload touch() {
       setupFrame.touch();

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

@@ -355,8 +355,7 @@ public Mono<RSocket> start() {
                           resumeToken,
                           metadataMimeType,
                           dataMimeType,
-                          setupPayload.sliceMetadata(),
-                          setupPayload.sliceData());
+                          setupPayload);
 
                   RSocket wrappedRSocketRequester = plugins.applyRequester(rSocketRequester);
 

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

@@ -205,7 +205,7 @@ private Mono<Void> handleFireAndForget(Payload payload) {
       return Mono.error(err);
     }
 
-    final int streamId = streamIdSupplier.nextStreamId();
+    final int streamId = streamIdSupplier.nextStreamId(receivers);
 
     return emptyUnicastMono()
         .doOnSubscribe(
@@ -233,7 +233,7 @@ private Mono<Payload> handleRequestResponse(final Payload payload) {
       return Mono.error(err);
     }
 
-    int streamId = streamIdSupplier.nextStreamId();
+    int streamId = streamIdSupplier.nextStreamId(receivers);
     final UnboundedProcessor<ByteBuf> sendProcessor = this.sendProcessor;
 
     UnicastMonoProcessor<Payload> receiver = UnicastMonoProcessor.create();
@@ -274,7 +274,7 @@ private Flux<Payload> handleRequestStream(final Payload payload) {
       return Flux.error(err);
     }
 
-    int streamId = streamIdSupplier.nextStreamId();
+    int streamId = streamIdSupplier.nextStreamId(receivers);
 
     final UnboundedProcessor<ByteBuf> sendProcessor = this.sendProcessor;
     final UnicastProcessor<Payload> receiver = UnicastProcessor.create();
@@ -328,7 +328,7 @@ private Flux<Payload> handleChannel(Flux<Payload> request) {
 
     final UnboundedProcessor<ByteBuf> sendProcessor = this.sendProcessor;
     final UnicastProcessor<Payload> receiver = UnicastProcessor.create();
-    final int streamId = streamIdSupplier.nextStreamId();
+    final int streamId = streamIdSupplier.nextStreamId(receivers);
 
     return receiver
         .doOnRequest(

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

@@ -13,18 +13,20 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package io.rsocket;
 
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import io.netty.util.collection.IntObjectMap;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 
 final class StreamIdSupplier {
+  private static final int MASK = 0x7FFFFFFF;
 
-  private static final AtomicIntegerFieldUpdater<StreamIdSupplier> STREAM_ID =
-      AtomicIntegerFieldUpdater.newUpdater(StreamIdSupplier.class, "streamId");
-  private volatile int streamId;
+  private static final AtomicLongFieldUpdater<StreamIdSupplier> STREAM_ID =
+      AtomicLongFieldUpdater.newUpdater(StreamIdSupplier.class, "streamId");
+  private volatile long streamId;
 
-  private StreamIdSupplier(int streamId) {
+  // Visible for testing
+  StreamIdSupplier(int streamId) {
     this.streamId = streamId;
   }
 
@@ -36,8 +38,12 @@ static StreamIdSupplier serverSupplier() {
     return new StreamIdSupplier(0);
   }
 
-  int nextStreamId() {
-    return STREAM_ID.addAndGet(this, 2);
+  int nextStreamId(IntObjectMap<?> streamIds) {
+    int streamId;
+    do {
+      streamId = (int) STREAM_ID.addAndGet(this, 2) & MASK;
+    } while (streamId == 0 || streamIds.containsKey(streamId));
+    return streamId;
   }
 
   boolean isBeforeOrCurrent(int streamId) {

rsocket-core/src/main/java/io/rsocket/buffer/Tuple3ByteBuf.java

@@ -144,7 +144,8 @@ public ByteBuffer[] _nioBuffers(int index, int length) {
                   new ByteBuffer[oneBuffer.length + twoBuffer.length + threeBuffer.length];
               System.arraycopy(oneBuffer, 0, results, 0, oneBuffer.length);
               System.arraycopy(twoBuffer, 0, results, oneBuffer.length, twoBuffer.length);
-              System.arraycopy(threeBuffer, 0, results, twoBuffer.length, threeBuffer.length);
+              System.arraycopy(
+                  threeBuffer, 0, results, oneBuffer.length + twoBuffer.length, threeBuffer.length);
               return results;
             } else {
               ByteBuffer[] results = new ByteBuffer[oneBuffer.length + twoBuffer.length];
@@ -167,7 +168,7 @@ public ByteBuffer[] _nioBuffers(int index, int length) {
             threeBuffer = three.nioBuffers(threeReadIndex, length);
             ByteBuffer[] results = new ByteBuffer[twoBuffer.length + threeBuffer.length];
             System.arraycopy(twoBuffer, 0, results, 0, twoBuffer.length);
-            System.arraycopy(threeBuffer, 0, results, threeBuffer.length, twoBuffer.length);
+            System.arraycopy(threeBuffer, 0, results, twoBuffer.length, threeBuffer.length);
             return results;
           } else {
             return twoBuffer;

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameReassembler.java

@@ -38,6 +38,9 @@
  *     and Reassembly</a>
  */
 final class FrameReassembler extends AtomicBoolean implements Disposable {
+
+  private static final long serialVersionUID = -4394598098863449055L;
+
   private static final Logger logger = LoggerFactory.getLogger(FrameReassembler.class);
 
   final IntObjectMap<ByteBuf> headers;