Replace use of deprecated Reactor Processor types

Signed-off-by: Rossen Stoyanchev <[email protected]>

Author: rstoyanchev

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

@@ -43,7 +43,7 @@
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoProcessor;
+import reactor.core.publisher.Sinks;
 import reactor.util.annotation.Nullable;
 
 /**
@@ -65,7 +65,7 @@ class RSocketRequester extends RequesterResponderSupport implements RSocket {
 
   @Nullable private final RequesterLeaseTracker requesterLeaseTracker;
   private final KeepAliveFramesAcceptor keepAliveFramesAcceptor;
-  private final MonoProcessor<Void> onClose;
+  private final Sinks.Empty<Void> onClose;
 
   RSocketRequester(
       DuplexConnection connection,
@@ -89,7 +89,7 @@ class RSocketRequester extends RequesterResponderSupport implements RSocket {
         requestInterceptorFunction);
 
     this.requesterLeaseTracker = requesterLeaseTracker;
-    this.onClose = MonoProcessor.create();
+    this.onClose = Sinks.empty();
 
     // DO NOT Change the order here. The Send processor must be subscribed to before receiving
     connection.onClose().subscribe(null, this::tryTerminateOnConnectionError, this::tryShutdown);
@@ -196,7 +196,7 @@ public boolean isDisposed() {
 
   @Override
   public Mono<Void> onClose() {
-    return onClose;
+    return onClose.asMono();
   }
 
   private void handleIncomingFrames(ByteBuf frame) {
@@ -356,9 +356,9 @@ private void terminate(Throwable e) {
     }
 
     if (e == CLOSED_CHANNEL_EXCEPTION) {
-      onClose.onComplete();
+      onClose.tryEmitEmpty();
     } else {
-      onClose.onError(e);
+      onClose.tryEmitError(e);
     }
   }
 }

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

@@ -1,17 +1,33 @@
+/*
+ * Copyright 2015-2021 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package io.rsocket.internal;
 
 import io.netty.buffer.ByteBuf;
 import io.rsocket.DuplexConnection;
+import reactor.core.Scannable;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoProcessor;
+import reactor.core.publisher.Sinks;
 
 public abstract class BaseDuplexConnection implements DuplexConnection {
-  protected MonoProcessor<Void> onClose = MonoProcessor.create();
+  protected Sinks.Empty<Void> onClose = Sinks.empty();
 
   protected UnboundedProcessor sender = new UnboundedProcessor();
 
   public BaseDuplexConnection() {
-    onClose.doFinally(s -> doOnClose()).subscribe();
+    onClose().doFinally(s -> doOnClose()).subscribe();
   }
 
   @Override
@@ -27,16 +43,17 @@ public void sendFrame(int streamId, ByteBuf frame) {
 
   @Override
   public final Mono<Void> onClose() {
-    return onClose;
+    return onClose.asMono();
   }
 
   @Override
   public final void dispose() {
-    onClose.onComplete();
+    onClose.tryEmitEmpty();
   }
 
   @Override
+  @SuppressWarnings("ConstantConditions")
   public final boolean isDisposed() {
-    return onClose.isDisposed();
+    return onClose.scan(Scannable.Attr.TERMINATED) || onClose.scan(Scannable.Attr.CANCELLED);
   }
 }

rsocket-core/src/main/java/io/rsocket/resume/InMemoryResumableFramesStore.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -28,8 +28,8 @@
 import reactor.core.CoreSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoProcessor;
 import reactor.core.publisher.Operators;
+import reactor.core.publisher.Sinks;
 
 /**
  * writes - n (where n is frequent, primary operation) reads - m (where m == KeepAliveFrequency)
@@ -40,7 +40,7 @@ public class InMemoryResumableFramesStore extends Flux<ByteBuf>
 
   private static final Logger logger = LoggerFactory.getLogger(InMemoryResumableFramesStore.class);
 
-  final MonoProcessor<Void> disposed = MonoProcessor.create();
+  final Sinks.Empty<Void> disposed = Sinks.empty();
   final ArrayList<ByteBuf> cachedFrames;
   final String tag;
   final int cacheLimit;
@@ -189,7 +189,7 @@ void resumeImplied() {
 
   @Override
   public Mono<Void> onClose() {
-    return disposed;
+    return disposed.asMono();
   }
 
   @Override
@@ -205,7 +205,7 @@ public void dispose() {
         }
         cachedFrames.clear();
       }
-      disposed.onComplete();
+      disposed.tryEmitEmpty();
     }
   }
 

rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2020 the original author or authors.
+ * Copyright 2015-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -30,9 +30,9 @@
 import org.slf4j.LoggerFactory;
 import reactor.core.CoreSubscriber;
 import reactor.core.Disposable;
+import reactor.core.Scannable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoProcessor;
 import reactor.core.publisher.Operators;
 import reactor.core.publisher.Sinks;
 
@@ -46,7 +46,7 @@ public class ResumableDuplexConnection extends Flux<ByteBuf>
 
   final UnboundedProcessor savableFramesSender;
   final Disposable framesSaverDisposable;
-  final MonoProcessor<Void> onClose;
+  final Sinks.Empty<Void> onClose;
   final SocketAddress remoteAddress;
   final Sinks.Many<Integer> onConnectionClosedSink;
 
@@ -72,7 +72,7 @@ public ResumableDuplexConnection(
     this.resumableFramesStore = resumableFramesStore;
     this.savableFramesSender = new UnboundedProcessor();
     this.framesSaverDisposable = resumableFramesStore.saveFrames(savableFramesSender).subscribe();
-    this.onClose = MonoProcessor.create();
+    this.onClose = Sinks.empty();
     this.remoteAddress = initialConnection.remoteAddress();
 
     ACTIVE_CONNECTION.lazySet(this, initialConnection);
@@ -164,17 +164,17 @@ public void sendErrorAndClose(RSocketErrorException rSocketErrorException) {
               framesSaverDisposable.dispose();
               savableFramesSender.dispose();
               onConnectionClosedSink.tryEmitComplete();
-              onClose.onError(t);
+              onClose.tryEmitError(t);
             },
             () -> {
               framesSaverDisposable.dispose();
               savableFramesSender.dispose();
               onConnectionClosedSink.tryEmitComplete();
               final Throwable cause = rSocketErrorException.getCause();
               if (cause == null) {
-                onClose.onComplete();
+                onClose.tryEmitEmpty();
               } else {
-                onClose.onError(cause);
+                onClose.tryEmitError(cause);
               }
             });
   }
@@ -191,7 +191,7 @@ public ByteBufAllocator alloc() {
 
   @Override
   public Mono<Void> onClose() {
-    return onClose;
+    return onClose.asMono();
   }
 
   @Override
@@ -210,12 +210,13 @@ public void dispose() {
     activeReceivingSubscriber.dispose();
     savableFramesSender.dispose();
     onConnectionClosedSink.tryEmitComplete();
-    onClose.onComplete();
+    onClose.tryEmitEmpty();
   }
 
   @Override
+  @SuppressWarnings("ConstantConditions")
   public boolean isDisposed() {
-    return onClose.isDisposed();
+    return onClose.scan(Scannable.Attr.TERMINATED) || onClose.scan(Scannable.Attr.CANCELLED);
   }
 
   @Override

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

@@ -1,6 +1,6 @@
 package io.rsocket.core;
 /*
- * Copyright 2015-2020 the original author or authors.
+ * Copyright 2015-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -57,8 +57,8 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Hooks;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoProcessor;
 import reactor.core.publisher.SignalType;
+import reactor.core.publisher.Sinks;
 import reactor.test.StepVerifier;
 import reactor.test.publisher.TestPublisher;
 import reactor.test.util.RaceTestUtils;
@@ -516,16 +516,17 @@ public static class ClientSocketRule extends AbstractSocketRule<RSocketRequester
 
     protected RSocketClient client;
     protected Runnable delayer;
-    protected MonoProcessor<RSocket> producer;
+    protected Sinks.One<RSocket> producer;
 
     @Override
     protected void init() {
       super.init();
-      delayer = () -> producer.onNext(socket);
-      producer = MonoProcessor.create();
+      delayer = () -> producer.tryEmitValue(socket);
+      producer = Sinks.one();
       client =
           new DefaultRSocketClient(
               producer
+                  .asMono()
                   .doOnCancel(() -> socket.dispose())
                   .doOnDiscard(Disposable.class, Disposable::dispose));
     }