migrates to the latest master updates

Signed-off-by: Oleh Dokuka <oleh.dokuka@icloud.com>
Signed-off-by: Oleh Dokuka <odokuka@vmware.com>
Signed-off-by: Oleh Dokuka <oleh.dokuka@icloud.com>

Author: OlegDokuka

rsocket-transport-aeron/build.gradle

@@ -17,8 +17,7 @@
 plugins {
     id 'java-library'
     id 'maven-publish'
-    id 'com.jfrog.artifactory'
-    id 'com.jfrog.bintray'
+    id 'signing'
     id "com.google.osdetector" version "1.4.0"
 }
 
@@ -35,7 +34,9 @@ dependencies {
     testImplementation 'org.mockito:mockito-junit-jupiter'
     testImplementation 'org.junit.jupiter:junit-jupiter-api'
     testImplementation 'org.junit.jupiter:junit-jupiter-params'
+    testImplementation 'org.slf4j:slf4j-api'
 
+    testImplementation 'ch.qos.logback:logback-classic'
     testRuntimeOnly 'ch.qos.logback:logback-classic'
     testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
 }

rsocket-transport-aeron/src/main/java/io/rsocket/transport/aeron/FluxReceiver.java

@@ -31,8 +31,8 @@
 import org.slf4j.LoggerFactory;
 import reactor.core.CoreSubscriber;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.MonoProcessor;
 import reactor.core.publisher.Operators;
+import reactor.core.publisher.Sinks;
 
 class FluxReceiver extends Flux<ByteBuf>
     implements org.reactivestreams.Subscription, ControlledFragmentHandler, Runnable {
@@ -44,7 +44,7 @@ class FluxReceiver extends Flux<ByteBuf>
   final Subscription subscription;
   final ControlledFragmentAssembler assembler;
   final EventLoop eventLoop;
-  final MonoProcessor<Void> onClose;
+  final Sinks.Empty<Void> onClose;
   final int effort;
 
   volatile long requested;
@@ -61,7 +61,7 @@ class FluxReceiver extends Flux<ByteBuf>
   int produced;
 
   public FluxReceiver(
-      MonoProcessor<Void> onClose, EventLoop eventLoop, Subscription subscription, int effort) {
+      Sinks.Empty<Void> onClose, EventLoop eventLoop, Subscription subscription, int effort) {
     this.onClose = onClose;
     this.eventLoop = eventLoop;
     this.subscription = subscription;
@@ -140,7 +140,7 @@ void drain(long n) {
             final NotConnectedException exception =
                 new NotConnectedException("Aeron Subscription has been closed unexpectedly");
             this.actual.onError(exception);
-            this.onClose.onError(exception);
+            this.onClose.tryEmitError(exception);
             return;
           }
 

rsocket-transport-aeron/src/main/java/io/rsocket/transport/aeron/MonoSendMany.java

@@ -28,17 +28,17 @@
 import reactor.core.Disposable;
 import reactor.core.Fuseable;
 import reactor.core.Fuseable.QueueSubscription;
-import reactor.core.publisher.MonoProcessor;
 import reactor.core.publisher.Operators;
+import reactor.core.publisher.Sinks;
 import reactor.util.context.Context;
 
 class MonoSendMany implements Disposable, CoreSubscriber<ByteBuf>, Runnable {
-  private static final Logger logger = LoggerFactory.getLogger(MonoSendMany.class);
+  static final Logger logger = LoggerFactory.getLogger(MonoSendMany.class);
 
   static final ThreadLocal<BufferClaim> BUFFER_CLAIMS = ThreadLocal.withInitial(BufferClaim::new);
   static final ThreadLocal<UnsafeBuffer> UNSAFE_BUFFER = ThreadLocal.withInitial(UnsafeBuffer::new);
 
-  final MonoProcessor<Void> onClose;
+  final Sinks.Empty<Void> onClose;
   final EventLoop eventLoop;
   final ExclusivePublication publication;
   final int effort;
@@ -59,7 +59,7 @@ class MonoSendMany implements Disposable, CoreSubscriber<ByteBuf>, Runnable {
   boolean cancelled;
 
   public MonoSendMany(
-      MonoProcessor<Void> onClose,
+      Sinks.Empty<Void> onClose,
       EventLoop eventLoop,
       ExclusivePublication publication,
       int effort,
@@ -130,7 +130,7 @@ void drain() {
   }
 
   void drainAsync() {
-    final MonoProcessor<Void> onClose = this.onClose;
+    final Sinks.Empty<Void> onClose = this.onClose;
     final ExclusivePublication publication = this.publication;
     final QueueSubscription<ByteBuf> qs = this.subscription;
     final EventLoop eventLoop = this.eventLoop;
@@ -159,7 +159,7 @@ void drainAsync() {
         qs.cancel();
 
         this.done = true;
-        onClose.onError(t);
+        onClose.tryEmitError(t);
         return;
       }
 
@@ -200,9 +200,9 @@ void drainAsync() {
 
             final Throwable t = this.t;
             if (t == null) {
-              onClose.onComplete();
+              onClose.tryEmitEmpty();
             } else {
-              onClose.onError(t);
+              onClose.tryEmitError(t);
             }
           }
           break;
@@ -217,7 +217,7 @@ void drainAsync() {
           buf.release();
 
           this.done = true;
-          onClose.onError(t);
+          onClose.tryEmitError(t);
           return;
         }
 

rsocket-transport-aeron/src/main/java/io/rsocket/transport/aeron/WrappedDirectBufferByteBuf.java

@@ -245,7 +245,7 @@ public int nioBufferCount() {
   public ByteBuffer nioBuffer(int index, int length) {
     final ByteBuffer buffer = directBuffer.byteBuffer();
     if (buffer != null) {
-      return buffer.duplicate().position(index).limit(index + length);
+      return (ByteBuffer) buffer.duplicate().position(index).limit(index + length);
     } else {
       final byte[] bytes = directBuffer.byteArray();
       return ByteBuffer.wrap(bytes, index, length);