provides prioritized sending for zero-stream frames; refactors benchmarks

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

Author: OlegDokuka

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

@@ -85,7 +85,7 @@ class RSocketResponder implements ResponderRSocket {
         .subscribe(null, this::handleSendProcessorError);
 
     Disposable receiveDisposable = connection.receive().subscribe(this::handleFrame, errorConsumer);
-    Disposable sendLeaseDisposable = leaseHandler.send(sendProcessor::onNext);
+    Disposable sendLeaseDisposable = leaseHandler.send(sendProcessor::onNextPrioritized);
 
     this.connection
         .onClose()

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

@@ -18,7 +18,6 @@
 
 import io.netty.util.ReferenceCounted;
 import io.rsocket.internal.jctools.queues.MpscUnboundedArrayQueue;
-import io.rsocket.internal.jctools.queues.SpscUnboundedArrayQueue;
 import java.util.Objects;
 import java.util.Queue;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
@@ -69,12 +68,12 @@ public final class UnboundedProcessor<T> extends FluxProcessor<T, T>
 
   public UnboundedProcessor() {
     this.queue = new MpscUnboundedArrayQueue<>(Queues.SMALL_BUFFER_SIZE);
-    this.priorityQueue = new SpscUnboundedArrayQueue<>(Queues.SMALL_BUFFER_SIZE);
+    this.priorityQueue = new MpscUnboundedArrayQueue<>(Queues.SMALL_BUFFER_SIZE);
   }
 
   @Override
   public int getBufferSize() {
-    return Queues.capacity(this.queue);
+    return Integer.MAX_VALUE;
   }
 
   @Override