more fixes

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

Author: OlegDokuka

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

@@ -411,7 +411,7 @@ private Mono<Void> handleMetadataPush(Payload payload) {
               MetadataPushFrameFlyweight.encode(allocator, payload.sliceMetadata().retain());
           payload.release();
 
-          sendProcessor.onNext(metadataPushFrame);
+          sendProcessor.onNextPrioritized(metadataPushFrame);
         });
   }
 

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

@@ -107,7 +107,7 @@ void drainRegular(Subscriber<? super T> a) {
           empty = t == null;
         }
 
-        if (checkTerminated(d, empty, a, q, pq)) {
+        if (checkTerminated(d, empty, a)) {
           return;
         }
 
@@ -121,7 +121,7 @@ void drainRegular(Subscriber<? super T> a) {
       }
 
       if (r == e) {
-        if (checkTerminated(done, q.isEmpty() && pq.isEmpty(), a, q, pq)) {
+        if (checkTerminated(done, q.isEmpty() && pq.isEmpty(), a)) {
           return;
         }
       }
@@ -140,9 +140,6 @@ void drainRegular(Subscriber<? super T> a) {
   void drainFused(Subscriber<? super T> a) {
     int missed = 1;
 
-    final Queue<T> q = queue;
-    final Queue<T> pq = priorityQueue;
-
     for (; ; ) {
 
       if (cancelled) {
@@ -201,7 +198,7 @@ public void drain() {
   }
 
   boolean checkTerminated(
-      boolean d, boolean empty, Subscriber<? super T> a, Queue<T> q, Queue<T> pq) {
+      boolean d, boolean empty, Subscriber<? super T> a) {
     if (cancelled) {
       clear();
       actual = null;
@@ -344,19 +341,12 @@ public void cancel() {
     }
   }
 
-  @Override
-  public T peek() {
-    if (!priorityQueue.isEmpty()) {
-      return priorityQueue.peek();
-    }
-    return queue.peek();
-  }
-
   @Override
   @Nullable
   public T poll() {
-    if (!priorityQueue.isEmpty()) {
-      return priorityQueue.poll();
+    Queue<T> pq = this.priorityQueue;
+    if (!pq.isEmpty()) {
+      return pq.poll();
     }
     return queue.poll();
   }