Fragmentation

build.gradle

@@ -40,6 +40,7 @@ subprojects {
         }
 
         dependencies {
+            dependency 'ch.qos.logback:logback-classic:1.2.3'
             dependency 'com.google.code.findbugs:jsr305:3.0.2'
             dependency 'com.netflix.spectator:spectator-api:0.63.0'
             dependency 'io.netty:netty-buffer:4.1.21.Final'
@@ -49,6 +50,7 @@ subprojects {
             dependency 'org.jctools:jctools-core:2.1.2'
             dependency 'org.mockito:mockito-core:2.16.0'
             dependency 'org.openjdk.jmh:jmh-core:1.20'
+            dependency 'org.slf4j:slf4j-api:1.7.25'
 
             dependencySet(group: 'org.junit.jupiter', version: '5.1.0') {
                 entry 'junit-jupiter-api'
@@ -66,11 +68,6 @@ subprojects {
                 entry 'jmh-core'
                 entry 'jmh-generator-annprocess'
             }
-
-            dependencySet(group: 'org.slf4j', version: '1.7.25') {
-                entry 'slf4j-api'
-                entry 'slf4j-nop'
-            }
         }
     }
 
@@ -86,6 +83,14 @@ subprojects {
             options.compilerArgs << '-Xlint:all,-overloads,-rawtypes,-unchecked'
         }
 
+        javadoc {
+            options.with {
+                links 'https://docs.oracle.com/javase/8/docs/api/'
+                links 'https://projectreactor.io/docs/core/release/api/'
+                links 'https://netty.io/4.1/api/'
+            }
+        }
+
         test {
             useJUnitPlatform()
         }

rsocket-core/build.gradle

@@ -27,6 +27,7 @@ plugins {
 dependencies {
     api 'io.netty:netty-buffer'
     api 'io.projectreactor:reactor-core'
+    api 'io.projectreactor.addons:reactor-extra'
 
     implementation 'com.google.code.findbugs:jsr305'
     implementation 'org.jctools:jctools-core'
@@ -37,6 +38,7 @@ dependencies {
     testImplementation 'org.junit.jupiter:junit-jupiter-api'
     testImplementation 'org.mockito:mockito-core'
 
+    testRuntimeOnly 'ch.qos.logback:logback-classic'
     testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
 
     // TODO: Remove after JUnit5 migration

rsocket-core/jmh.gradle

@@ -16,11 +16,9 @@
 
 dependencies {
     jmh configurations.api
+    jmh configurations.implementation
     jmh 'org.openjdk.jmh:jmh-core'
     jmh 'org.openjdk.jmh:jmh-generator-annprocess'
-
-    jmhRuntime 'org.openjdk.jmh:jmh-core'
-    jmhRuntime 'org.slf4j:slf4j-nop'
 }
 
 jmhCompileGeneratedClasses.enabled = false
@@ -36,7 +34,6 @@ jmh {
 
 jmhJar {
     from project.configurations.jmh
-    from project.configurations.jmhRuntime
 }
 
 tasks.jmh.finalizedBy tasks.jmhReport

rsocket-core/src/jmh/java/io/rsocket/fragmentation/FragmentationPerformanceTest.java

@@ -0,0 +1,152 @@
+/*
+ * Copyright 2015-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.rsocket.fragmentation;
+
+import static io.netty.buffer.UnpooledByteBufAllocator.DEFAULT;
+import static io.rsocket.framing.RequestResponseFrame.createRequestResponseFrame;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import io.rsocket.framing.Frame;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+import reactor.core.publisher.SynchronousSink;
+import reactor.util.context.Context;
+
+@BenchmarkMode(Mode.Throughput)
+@Fork(
+  value = 1 // , jvmArgsAppend = {"-Dio.netty.leakDetection.level=advanced"}
+)
+@Warmup(iterations = 10)
+@Measurement(iterations = 10_000)
+@State(Scope.Thread)
+public class FragmentationPerformanceTest {
+
+  @Benchmark
+  public void largeFragmentation(Input input) {
+    Frame frame =
+        input.largeFragmenter.fragment(input.largeFrame).doOnNext(Frame::dispose).blockLast();
+
+    input.bh.consume(frame);
+  }
+
+  @Benchmark
+  public void largeReassembly(Input input) {
+    input.largeFrames.forEach(frame -> input.reassembler.reassemble(frame, input.sink));
+
+    input.bh.consume(input.sink.next);
+  }
+
+  @Benchmark
+  public void smallFragmentation(Input input) {
+    Frame frame =
+        input.smallFragmenter.fragment(input.smallFrame).doOnNext(Frame::dispose).blockLast();
+
+    input.bh.consume(frame);
+  }
+
+  @Benchmark
+  public void smallReassembly(Input input) {
+    input.smallFrames.forEach(frame -> input.reassembler.reassemble(frame, input.sink));
+
+    input.bh.consume(input.sink.next);
+  }
+
+  @State(Scope.Benchmark)
+  public static class Input {
+
+    Blackhole bh;
+
+    FrameFragmenter largeFragmenter;
+
+    Frame largeFrame;
+
+    List<Frame> largeFrames;
+
+    FrameReassembler reassembler = FrameReassembler.createFrameReassembler(DEFAULT);
+
+    MockSynchronousSink<Frame> sink;
+
+    FrameFragmenter smallFragmenter;
+
+    Frame smallFrame;
+
+    List<Frame> smallFrames;
+
+    @Setup
+    public void setup(Blackhole bh) {
+      this.bh = bh;
+
+      sink = new MockSynchronousSink<>();
+
+      largeFrame =
+          createRequestResponseFrame(
+              DEFAULT, false, getRandomByteBuf(1 << 18), getRandomByteBuf(1 << 18));
+
+      smallFrame =
+          createRequestResponseFrame(DEFAULT, false, getRandomByteBuf(16), getRandomByteBuf(16));
+
+      largeFragmenter = new FrameFragmenter(DEFAULT, 1024);
+      smallFragmenter = new FrameFragmenter(ByteBufAllocator.DEFAULT, 2);
+
+      largeFrames = largeFragmenter.fragment(largeFrame).collectList().block();
+      smallFrames = smallFragmenter.fragment(smallFrame).collectList().block();
+    }
+
+    private static ByteBuf getRandomByteBuf(int size) {
+      byte[] bytes = new byte[size];
+      ThreadLocalRandom.current().nextBytes(bytes);
+      return Unpooled.wrappedBuffer(bytes);
+    }
+  }
+
+  static final class MockSynchronousSink<T> implements SynchronousSink<T> {
+
+    Throwable error;
+
+    T next;
+
+    @Override
+    public void complete() {}
+
+    @Override
+    public Context currentContext() {
+      return null;
+    }
+
+    @Override
+    public void error(Throwable e) {
+      this.error = e;
+    }
+
+    @Override
+    public void next(T t) {
+      this.next = t;
+    }
+  }
+}

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

@@ -23,6 +23,7 @@
 import io.netty.util.AbstractReferenceCounted;
 import io.rsocket.Frame.Setup;
 import io.rsocket.frame.SetupFrameFlyweight;
+import io.rsocket.framing.FrameType;
 
 /**
  * Exposed to server for determination of RequestHandler based on mime types and SETUP metadata/data

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

@@ -33,11 +33,12 @@ public interface DuplexConnection extends Availability, Closeable {
    *
    * The passed {@code Publisher} must
    *
-   * @param frame Stream of {@code Frame}s to send on the connection.
+   * @param frames Stream of {@code Frame}s to send on the connection.
    * @return {@code Publisher} that completes when all the frames are written on the connection
    *     successfully and errors when it fails.
+   * @throws NullPointerException if {@code frames} is {@code null}
    */
-  Mono<Void> send(Publisher<Frame> frame);
+  Mono<Void> send(Publisher<Frame> frames);
 
   /**
    * Sends a single {@code Frame} on this connection and returns the {@code Publisher} representing