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;

rsocket-core/src/main/java/io/rsocket/frame/SetupFrameFlyweight.java

@@ -4,6 +4,7 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
+import io.rsocket.Payload;
 import java.nio.charset.StandardCharsets;
 
 public class SetupFrameFlyweight {
@@ -27,13 +28,12 @@ public class SetupFrameFlyweight {
 
   public static ByteBuf encode(
       final ByteBufAllocator allocator,
-      boolean lease,
+      final boolean lease,
       final int keepaliveInterval,
       final int maxLifetime,
       final String metadataMimeType,
       final String dataMimeType,
-      final ByteBuf metadata,
-      final ByteBuf data) {
+      final Payload setupPayload) {
     return encode(
         allocator,
         lease,
@@ -42,20 +42,21 @@ public static ByteBuf encode(
         Unpooled.EMPTY_BUFFER,
         metadataMimeType,
         dataMimeType,
-        metadata,
-        data);
+        setupPayload);
   }
 
   public static ByteBuf encode(
       final ByteBufAllocator allocator,
-      boolean lease,
+      final boolean lease,
       final int keepaliveInterval,
       final int maxLifetime,
       final ByteBuf resumeToken,
       final String metadataMimeType,
       final String dataMimeType,
-      final ByteBuf metadata,
-      final ByteBuf data) {
+      final Payload setupPayload) {
+
+    ByteBuf metadata = setupPayload.hasMetadata() ? setupPayload.sliceMetadata() : null;
+    ByteBuf data = setupPayload.sliceData();
 
     int flags = 0;
 

rsocket-core/src/main/java/io/rsocket/internal/SynchronizedIntObjectHashMap.java

@@ -189,7 +189,7 @@ public synchronized boolean isEmpty() {
 
   @Override
   public synchronized void clear() {
-    Arrays.fill(keys, (int) 0);
+    Arrays.fill(keys, 0);
     Arrays.fill(values, null);
     size = 0;
   }
@@ -331,7 +331,7 @@ public synchronized Set<Entry<Integer, V>> entrySet() {
   }
 
   private int objectToKey(Object key) {
-    return (int) ((Integer) key).intValue();
+    return ((Integer) key).intValue();
   }
 
   /**
@@ -369,7 +369,7 @@ private int hashIndex(int key) {
 
   /** Returns the hash code for the key. */
   private static int hashCode(int key) {
-    return (int) key;
+    return key;
   }
 
   /** Get the next sequential index after {@code index} and wraps if necessary. */

rsocket-core/src/main/java/io/rsocket/internal/UnicastMonoProcessor.java

@@ -2,7 +2,6 @@
 
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
-import java.util.function.LongSupplier;
 import java.util.stream.Stream;
 import org.reactivestreams.Processor;
 import org.reactivestreams.Subscription;
@@ -17,12 +16,7 @@
 import reactor.util.function.Tuple2;
 
 public class UnicastMonoProcessor<O> extends Mono<O>
-    implements Processor<O, O>,
-        CoreSubscriber<O>,
-        Disposable,
-        Subscription,
-        Scannable,
-        LongSupplier {
+    implements Processor<O, O>, CoreSubscriber<O>, Disposable, Subscription, Scannable {
 
   @SuppressWarnings("rawtypes")
   static final AtomicIntegerFieldUpdater<UnicastMonoProcessor> ONCE =
@@ -87,11 +81,6 @@ public Stream<Tuple2<String, String>> tags() {
     return processor.tags();
   }
 
-  @Override
-  public long getAsLong() {
-    return processor.getAsLong();
-  }
-
   @Override
   public void onSubscribe(Subscription s) {
     processor.onSubscribe(s);

rsocket-core/src/main/java/io/rsocket/metadata/WellKnownMimeType.java

@@ -67,6 +67,9 @@ public enum WellKnownMimeType {
   VIDEO_H264("video/H264", (byte) 0x23),
   VIDEO_H265("video/H265", (byte) 0x24),
   VIDEO_VP8("video/VP8", (byte) 0x25),
+  APPLICATION_HESSIAN("application/x-hessian", (byte) 0x26),
+  APPLICATION_JAVA_OBJECT("application/x-java-object", (byte) 0x27),
+  APPLICATION_CLOUDEVENTS_JSON("application/cloudevents+json", (byte) 0x28),
 
   // ... reserved for future use ...