Merge #1.0.x into master

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

Author: OlegDokuka

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

@@ -170,19 +170,19 @@ public T block(@Nullable Duration timeout) {
         delay = System.nanoTime() + timeout.toNanos();
       }
       for (; ; ) {
-        BiConsumer<T, Throwable>[] inners = this.subscribers;
+        subscribers = this.subscribers;
 
-        if (inners == READY) {
+        if (subscribers == READY) {
           final T value = this.value;
           if (value != null) {
             return value;
           } else {
             // value == null means racing between invalidate and this block
             // thus, we have to update the state again and see what happened
-            inners = this.subscribers;
+            subscribers = this.subscribers;
           }
         }
-        if (inners == TERMINATED) {
+        if (subscribers == TERMINATED) {
           RuntimeException re = Exceptions.propagate(this.t);
           re = Exceptions.addSuppressed(re, new Exception("Terminated with an error"));
           throw re;
@@ -191,6 +191,12 @@ public T block(@Nullable Duration timeout) {
           throw new IllegalStateException("Timeout on Mono blocking read");
         }
 
+        // connect again since invalidate() has happened in between
+        if (subscribers == EMPTY_UNSUBSCRIBED
+            && SUBSCRIBERS.compareAndSet(this, EMPTY_UNSUBSCRIBED, EMPTY_SUBSCRIBED)) {
+          this.doSubscribe();
+        }
+
         Thread.sleep(1);
       }
     } catch (InterruptedException ie) {
@@ -203,6 +209,7 @@ public T block(@Nullable Duration timeout) {
   @SuppressWarnings("unchecked")
   final void terminate(Throwable t) {
     if (isDisposed()) {
+      Operators.onErrorDropped(t, Context.empty());
       return;
     }
 

rsocket-core/src/main/java/io/rsocket/loadbalance/ResolvingOperator.java

@@ -18,14 +18,10 @@
 import java.time.Duration;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
-import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.function.BiConsumer;
-import org.reactivestreams.Subscription;
-import reactor.core.CoreSubscriber;
 import reactor.core.Disposable;
 import reactor.core.Exceptions;
-import reactor.core.Scannable;
 import reactor.core.publisher.Operators;
 import reactor.util.annotation.Nullable;
 import reactor.util.context.Context;
@@ -170,19 +166,19 @@ public T block(@Nullable Duration timeout) {
         delay = System.nanoTime() + timeout.toNanos();
       }
       for (; ; ) {
-        BiConsumer<T, Throwable>[] inners = this.subscribers;
+        subscribers = this.subscribers;
 
-        if (inners == READY) {
+        if (subscribers == READY) {
           final T value = this.value;
           if (value != null) {
             return value;
           } else {
             // value == null means racing between invalidate and this block
             // thus, we have to update the state again and see what happened
-            inners = this.subscribers;
+            subscribers = this.subscribers;
           }
         }
-        if (inners == TERMINATED) {
+        if (subscribers == TERMINATED) {
           RuntimeException re = Exceptions.propagate(this.t);
           re = Exceptions.addSuppressed(re, new Exception("Terminated with an error"));
           throw re;
@@ -191,6 +187,12 @@ public T block(@Nullable Duration timeout) {
           throw new IllegalStateException("Timeout on Mono blocking read");
         }
 
+        // connect again since invalidate() has happened in between
+        if (subscribers == EMPTY_UNSUBSCRIBED
+            && SUBSCRIBERS.compareAndSet(this, EMPTY_UNSUBSCRIBED, EMPTY_SUBSCRIBED)) {
+          this.doSubscribe();
+        }
+
         Thread.sleep(1);
       }
     } catch (InterruptedException ie) {
@@ -203,6 +205,7 @@ public T block(@Nullable Duration timeout) {
   @SuppressWarnings("unchecked")
   final void terminate(Throwable t) {
     if (isDisposed()) {
+      Operators.onErrorDropped(t, Context.empty());
       return;
     }
 
@@ -390,175 +393,4 @@ final void remove(BiConsumer<T, Throwable> ps) {
       }
     }
   }
-
-  abstract static class DeferredResolution<T, R>
-      implements CoreSubscriber<T>, Subscription, Scannable, BiConsumer<R, Throwable> {
-
-    final ResolvingOperator<R> parent;
-    final CoreSubscriber<? super T> actual;
-
-    volatile long requested;
-
-    @SuppressWarnings("rawtypes")
-    static final AtomicLongFieldUpdater<DeferredResolution> REQUESTED =
-        AtomicLongFieldUpdater.newUpdater(DeferredResolution.class, "requested");
-
-    static final long STATE_SUBSCRIBED = -1;
-    static final long STATE_CANCELLED = Long.MIN_VALUE;
-
-    Subscription s;
-    boolean done;
-
-    DeferredResolution(ResolvingOperator<R> parent, CoreSubscriber<? super T> actual) {
-      this.parent = parent;
-      this.actual = actual;
-    }
-
-    @Override
-    public final Context currentContext() {
-      return this.actual.currentContext();
-    }
-
-    @Nullable
-    @Override
-    public Object scanUnsafe(Attr key) {
-      long state = this.requested;
-
-      if (key == Attr.PARENT) {
-        return this.s;
-      }
-      if (key == Attr.ACTUAL) {
-        return this.parent;
-      }
-      if (key == Attr.TERMINATED) {
-        return this.done;
-      }
-      if (key == Attr.CANCELLED) {
-        return state == STATE_CANCELLED;
-      }
-
-      return null;
-    }
-
-    @Override
-    public final void onSubscribe(Subscription s) {
-      final long state = this.requested;
-      Subscription a = this.s;
-      if (state == STATE_CANCELLED) {
-        s.cancel();
-        return;
-      }
-      if (a != null) {
-        s.cancel();
-        return;
-      }
-
-      long r;
-      long accumulated = 0;
-      for (; ; ) {
-        r = this.requested;
-
-        if (r == STATE_CANCELLED || r == STATE_SUBSCRIBED) {
-          s.cancel();
-          return;
-        }
-
-        this.s = s;
-
-        long toRequest = r - accumulated;
-        if (toRequest > 0) { // if there is something,
-          s.request(toRequest); // then we do a request on the given subscription
-        }
-        accumulated = r;
-
-        if (REQUESTED.compareAndSet(this, r, STATE_SUBSCRIBED)) {
-          return;
-        }
-      }
-    }
-
-    @Override
-    public final void onNext(T payload) {
-      this.actual.onNext(payload);
-    }
-
-    @Override
-    public final void onError(Throwable t) {
-      if (this.done) {
-        Operators.onErrorDropped(t, this.actual.currentContext());
-        return;
-      }
-
-      this.done = true;
-      this.actual.onError(t);
-    }
-
-    @Override
-    public final void onComplete() {
-      if (this.done) {
-        return;
-      }
-
-      this.done = true;
-      this.actual.onComplete();
-    }
-
-    @Override
-    public void request(long n) {
-      if (Operators.validate(n)) {
-        long r = this.requested; // volatile read beforehand
-
-        if (r > STATE_SUBSCRIBED) { // works only in case onSubscribe has not happened
-          long u;
-          for (; ; ) { // normal CAS loop with overflow protection
-            if (r == Long.MAX_VALUE) {
-              // if r == Long.MAX_VALUE then we dont care and we can loose this
-              // request just in case of racing
-              return;
-            }
-            u = Operators.addCap(r, n);
-            if (REQUESTED.compareAndSet(this, r, u)) {
-              // Means increment happened before onSubscribe
-              return;
-            } else {
-              // Means increment happened after onSubscribe
-
-              // update new state to see what exactly happened (onSubscribe |cancel | requestN)
-              r = this.requested;
-
-              // check state (expect -1 | -2 to exit, otherwise repeat)
-              if (r < 0) {
-                break;
-              }
-            }
-          }
-        }
-
-        if (r == STATE_CANCELLED) { // if canceled, just exit
-          return;
-        }
-
-        // if onSubscribe -> subscription exists (and we sure of that because volatile read
-        // after volatile write) so we can execute requestN on the subscription
-        this.s.request(n);
-      }
-    }
-
-    public boolean isCancelled() {
-      return this.requested == STATE_CANCELLED;
-    }
-
-    public void cancel() {
-      long state = REQUESTED.getAndSet(this, STATE_CANCELLED);
-      if (state == STATE_CANCELLED) {
-        return;
-      }
-
-      if (state == STATE_SUBSCRIBED) {
-        this.s.cancel();
-      } else {
-        this.parent.remove(this);
-      }
-    }
-  }
 }

rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java

@@ -105,7 +105,6 @@
 import reactor.core.publisher.Hooks;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
-import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 import reactor.test.publisher.TestPublisher;
 import reactor.test.util.RaceTestUtils;
@@ -1171,12 +1170,8 @@ public void shouldTerminateAllStreamsIfThereRacingBetweenDisposeAndRequests(
       Publisher<?> publisher2 = interaction2.apply(rule, payload2);
       RaceTestUtils.race(
           () -> rule.socket.dispose(),
-          () ->
-              RaceTestUtils.race(
-                  () -> publisher1.subscribe(assertSubscriber1),
-                  () -> publisher2.subscribe(assertSubscriber2),
-                  Schedulers.parallel()),
-          Schedulers.parallel());
+          () -> publisher1.subscribe(assertSubscriber1),
+          () -> publisher2.subscribe(assertSubscriber2));
 
       assertSubscriber1.await().assertTerminated();
       if (interactionType1 != REQUEST_FNF && interactionType1 != METADATA_PUSH) {