reduces maintenance complexity

this PR
 1. removes set of unused classes
 2. eliminate usage of custom publishers implementation if favor of better maintainability but sacrificing performance (1.1 will revise that)

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

Author: OlegDokuka

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

@@ -45,21 +45,18 @@
 import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.internal.SynchronizedIntObjectHashMap;
 import io.rsocket.internal.UnboundedProcessor;
-import io.rsocket.internal.UnicastMonoEmpty;
-import io.rsocket.internal.UnicastMonoProcessor;
 import io.rsocket.keepalive.KeepAliveFramesAcceptor;
 import io.rsocket.keepalive.KeepAliveHandler;
 import io.rsocket.keepalive.KeepAliveSupport;
 import io.rsocket.lease.RequesterLeaseHandler;
-import io.rsocket.util.MonoLifecycleHandler;
 import java.nio.channels.ClosedChannelException;
+import java.util.Queue;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.function.Consumer;
 import java.util.function.LongConsumer;
 import java.util.function.Supplier;
-import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import org.reactivestreams.Processor;
 import org.reactivestreams.Publisher;
@@ -92,6 +89,8 @@ class RSocketRequester implements RSocket {
         }
       };
 
+  private static final Queue<Void> EMPTY_QUEUE = Queues.<Void>empty().get();
+
   static {
     CLOSED_CHANNEL_EXCEPTION.setStackTrace(new StackTraceElement[0]);
   }
@@ -212,14 +211,16 @@ private Mono<Void> handleFireAndForget(Payload payload) {
 
     final int streamId = streamIdSupplier.nextStreamId(receivers);
 
-    return UnicastMonoEmpty.newInstance(
-        () -> {
-          ByteBuf requestFrame =
-              RequestFireAndForgetFrameFlyweight.encodeReleasingPayload(
-                  allocator, streamId, payload);
+    return UnicastProcessor.create(EMPTY_QUEUE)
+        .doOnSubscribe(
+            (__) -> {
+              ByteBuf requestFrame =
+                  RequestFireAndForgetFrameFlyweight.encodeReleasingPayload(
+                      allocator, streamId, payload);
 
-          sendProcessor.onNext(requestFrame);
-        });
+              sendProcessor.onNext(requestFrame);
+            })
+        .then();
   }
 
   private Mono<Payload> handleRequestResponse(final Payload payload) {
@@ -236,34 +237,28 @@ private Mono<Payload> handleRequestResponse(final Payload payload) {
 
     int streamId = streamIdSupplier.nextStreamId(receivers);
     final UnboundedProcessor<ByteBuf> sendProcessor = this.sendProcessor;
-
-    UnicastMonoProcessor<Payload> receiver =
-        UnicastMonoProcessor.create(
-            new MonoLifecycleHandler<Payload>() {
-              @Override
-              public void doOnSubscribe() {
-                final ByteBuf requestFrame =
-                    RequestResponseFrameFlyweight.encodeReleasingPayload(
-                        allocator, streamId, payload);
-
-                sendProcessor.onNext(requestFrame);
-              }
-
-              @Override
-              public void doOnTerminal(
-                  @Nonnull SignalType signalType,
-                  @Nullable Payload element,
-                  @Nullable Throwable e) {
-                if (signalType == SignalType.CANCEL) {
-                  sendProcessor.onNext(CancelFrameFlyweight.encode(allocator, streamId));
-                }
-                removeStreamReceiver(streamId);
-              }
-            });
+    final UnicastProcessor<Payload> receiver = UnicastProcessor.create(Queues.<Payload>one().get());
 
     receivers.put(streamId, receiver);
 
-    return receiver.doOnDiscard(ReferenceCounted.class, DROPPED_ELEMENTS_CONSUMER);
+    return receiver
+        .as(Mono::fromDirect)
+        .doOnSubscribe(
+            (__) -> {
+              ByteBuf requestFrame =
+                  RequestResponseFrameFlyweight.encodeReleasingPayload(
+                      allocator, streamId, payload);
+
+              sendProcessor.onNext(requestFrame);
+            })
+        .doFinally(
+            signalType -> {
+              if (signalType == SignalType.CANCEL) {
+                sendProcessor.onNext(CancelFrameFlyweight.encode(allocator, streamId));
+              }
+              removeStreamReceiver(streamId);
+            })
+        .doOnDiscard(ReferenceCounted.class, DROPPED_ELEMENTS_CONSUMER);
   }
 
   private Flux<Payload> handleRequestStream(final Payload payload) {
@@ -522,13 +517,15 @@ private Mono<Void> handleMetadataPush(Payload payload) {
       return Mono.error(new IllegalArgumentException(INVALID_PAYLOAD_ERROR_MESSAGE));
     }
 
-    return UnicastMonoEmpty.newInstance(
-        () -> {
-          ByteBuf metadataPushFrame =
-              MetadataPushFrameFlyweight.encodeReleasingPayload(allocator, payload);
+    return UnicastProcessor.create(EMPTY_QUEUE)
+        .doOnSubscribe(
+            (__) -> {
+              ByteBuf metadataPushFrame =
+                  MetadataPushFrameFlyweight.encodeReleasingPayload(allocator, payload);
 
-          sendProcessor.onNextPrioritized(metadataPushFrame);
-        });
+              sendProcessor.onNextPrioritized(metadataPushFrame);
+            })
+        .then();
   }
 
   @Nullable