provides more tests

part of the tests are on racing (ignored for now)
another few on verification that elements are discarded properly

Signed-off-by: Oleh Dokuka <[email protected]>

Author: OlegDokuka

rsocket-core/build.gradle

@@ -46,4 +46,23 @@ dependencies {
     testRuntimeOnly 'org.junit.vintage:junit-vintage-engine'
 }
 
+if (JavaVersion.current().java9Compatible) {
+    sourceSets {
+        java9impl.java.srcDirs = ['src/test/java9impl']
+    }
+
+    dependencies {
+        testCompile sourceSets.java9impl.output
+    }
+}
+else {
+    sourceSets {
+        java8impl.java.srcDirs = ['src/test/java8impl']
+    }
+
+    dependencies {
+        testCompile sourceSets.java8impl.output
+    }
+}
+
 description = "Core functionality for the RSocket library"

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

@@ -16,7 +16,6 @@
 
 package io.rsocket.core;
 
-import static io.rsocket.core.LimitRateOperator.applyLimitRateOperator;
 import static io.rsocket.core.PayloadValidationUtils.INVALID_PAYLOAD_ERROR_MESSAGE;
 import static io.rsocket.keepalive.KeepAliveSupport.ClientKeepAliveSupport;
 import static io.rsocket.keepalive.KeepAliveSupport.KeepAlive;
@@ -281,8 +280,7 @@ private Flux<Payload> handleRequestStream(final Payload payload) {
     int streamId = streamIdSupplier.nextStreamId(receivers);
 
     final UnboundedProcessor<ByteBuf> sendProcessor = this.sendProcessor;
-    final UnicastProcessor<Payload> receiver =
-        UnicastProcessor.create(new CleanOnClearQueueDecorator(Queues.<Payload>unbounded().get()));
+    final UnicastProcessor<Payload> receiver = UnicastProcessor.create();
     final AtomicBoolean payloadReleasedFlag = new AtomicBoolean(false);
 
     receivers.put(streamId, receiver);
@@ -362,8 +360,7 @@ private Flux<? extends Payload> handleChannel(Payload initialPayload, Flux<Paylo
     final AtomicBoolean payloadReleasedFlag = new AtomicBoolean(false);
     final int streamId = streamIdSupplier.nextStreamId(receivers);
 
-    final UnicastProcessor<Payload> receiver =
-        UnicastProcessor.create(new CleanOnClearQueueDecorator(Queues.<Payload>unbounded().get()));
+    final UnicastProcessor<Payload> receiver = UnicastProcessor.create();
     final BaseSubscriber<Payload> upstreamSubscriber =
         new BaseSubscriber<Payload>() {
 
@@ -432,7 +429,7 @@ public void accept(long n) {
                   receivers.put(streamId, receiver);
 
                   inboundFlux
-                      .transform(f -> applyLimitRateOperator(f, Queues.SMALL_BUFFER_SIZE))
+                      .limitRate(Queues.SMALL_BUFFER_SIZE)
                       .doOnDiscard(ReferenceCounted.class, DROPPED_ELEMENTS_CONSUMER)
                       .subscribe(upstreamSubscriber);
                   if (!payloadReleasedFlag.getAndSet(true)) {

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

@@ -16,7 +16,6 @@
 
 package io.rsocket.core;
 
-import static io.rsocket.core.LimitRateOperator.applyLimitRateOperator;
 import static io.rsocket.core.PayloadValidationUtils.INVALID_PAYLOAD_ERROR_MESSAGE;
 
 import io.netty.buffer.ByteBuf;
@@ -476,14 +475,13 @@ protected void hookFinally(SignalType type) {
 
     sendingSubscriptions.put(streamId, subscriber);
     response
-        .transform(f -> applyLimitRateOperator(f, Queues.SMALL_BUFFER_SIZE))
+        .limitRate(Queues.SMALL_BUFFER_SIZE)
         .doOnDiscard(ReferenceCounted.class, DROPPED_ELEMENTS_CONSUMER)
         .subscribe(subscriber);
   }
 
   private void handleChannel(int streamId, Payload payload, int initialRequestN) {
-    UnicastProcessor<Payload> frames =
-        UnicastProcessor.create(new CleanOnClearQueueDecorator(Queues.<Payload>unbounded().get()));
+    UnicastProcessor<Payload> frames = UnicastProcessor.create();
     channelProcessors.put(streamId, frames);
 
     Flux<Payload> payloads =

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

@@ -2,15 +2,35 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.util.IllegalReferenceCountException;
+import io.netty.util.ReferenceCounted;
 import io.rsocket.Payload;
 import io.rsocket.frame.*;
 import io.rsocket.util.ByteBufPayload;
+import reactor.core.publisher.Hooks;
 
 /**
  * Frame decoder that decodes a frame to a payload without copying. The caller is responsible for
  * for releasing the payload to free memory when they no long need it.
  */
 public class ZeroCopyPayloadDecoder implements PayloadDecoder {
+
+  static {
+    Hooks.onNextDropped(
+        o -> {
+          if (o instanceof ReferenceCounted) {
+            ReferenceCounted referenceCounted = (ReferenceCounted) o;
+            if (referenceCounted.refCnt() > 0) {
+              try {
+                referenceCounted.release();
+              } catch (IllegalReferenceCountException e) {
+                // ignored
+              }
+            }
+          }
+        });
+  }
+
   @Override
   public Payload apply(ByteBuf byteBuf) {
     ByteBuf m;