Use ReferenceCountUtil.safeRelease instead of release (#505)

Author: Ryland Degnan

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

@@ -25,6 +25,7 @@
 import reactor.core.CoreSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Operators;
+import reactor.core.publisher.UnicastProcessor;
 
 public final class SwitchTransform<T, R> extends Flux<R> {
 
@@ -39,7 +40,7 @@ public SwitchTransform(
 
   @Override
   public void subscribe(CoreSubscriber<? super R> actual) {
-    Flux.from(source).subscribe(new SwitchTransformSubscriber<>(actual, transformer));
+    source.subscribe(new SwitchTransformSubscriber<>(actual, transformer));
   }
 
   static final class SwitchTransformSubscriber<T, R>
@@ -50,7 +51,7 @@ static final class SwitchTransformSubscriber<T, R>
 
     final CoreSubscriber<? super R> actual;
     final BiFunction<T, Flux<T>, Publisher<? extends R>> transformer;
-    final UnboundedProcessor<T> processor = new UnboundedProcessor<>();
+    final UnicastProcessor<T> processor = UnicastProcessor.create();
     Subscription s;
     volatile int once;
 
@@ -76,10 +77,10 @@ public void onNext(T t) {
           Publisher<? extends R> result =
               Objects.requireNonNull(
                   transformer.apply(t, processor), "The transformer returned a null value");
-          Flux.from(result).subscribe(actual);
+          result.subscribe(actual);
         } catch (Throwable e) {
           onError(Operators.onOperatorError(s, e, t, actual.currentContext()));
-          ReferenceCountUtil.release(t);
+          ReferenceCountUtil.safeRelease(t);
           return;
         }
       }

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

@@ -158,7 +158,7 @@ boolean checkTerminated(boolean d, boolean empty, Subscriber<? super T> a, Queue
       while (!q.isEmpty()) {
         T t = q.poll();
         if (t != null) {
-          ReferenceCountUtil.release(t);
+          ReferenceCountUtil.safeRelease(t);
         }
       }
       actual = null;
@@ -202,15 +202,15 @@ public Context currentContext() {
   public void onNext(T t) {
     if (done || cancelled) {
       Operators.onNextDropped(t, currentContext());
-      ReferenceCountUtil.release(t);
+      ReferenceCountUtil.safeRelease(t);
       return;
     }
 
     if (!queue.offer(t)) {
       Throwable ex =
           Operators.onOperatorError(null, Exceptions.failWithOverflow(), t, currentContext());
       onError(Operators.onOperatorError(null, ex, t, currentContext()));
-      ReferenceCountUtil.release(t);
+      ReferenceCountUtil.safeRelease(t);
       return;
     }
     drain();
@@ -301,7 +301,7 @@ public void clear() {
     while (!queue.isEmpty()) {
       T t = queue.poll();
       if (t != null) {
-        ReferenceCountUtil.release(t);
+        ReferenceCountUtil.safeRelease(t);
       }
     }
   }