Upgrade to Reactor 2022.0.0-M1

[resolves #503]

Signed-off-by: Mark Paluch <[email protected]>

Author: mp911de

pom.xml

@@ -48,7 +48,7 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
         <r2dbc-spi.version>1.0.0.BUILD-SNAPSHOT</r2dbc-spi.version>
-        <reactor.version>2020.0.17</reactor.version>
+        <reactor.version>2022.0.0-M1</reactor.version>
         <scram-client.version>2.1</scram-client.version>
         <spring-framework.version>5.3.16</spring-framework.version>
         <testcontainers.version>1.16.3</testcontainers.version>
@@ -648,6 +648,11 @@
                 <enabled>true</enabled>
             </snapshots>
         </repository>
+
+        <repository>
+            <id>spring-libs-milestone</id>
+            <url>https://repo.spring.io/libs-milestone</url>
+        </repository>
     </repositories>
 
 </project>

src/main/java/io/r2dbc/postgresql/PostgresqlConnection.java

@@ -38,10 +38,10 @@
 import io.r2dbc.spi.TransactionDefinition;
 import io.r2dbc.spi.ValidationDepth;
 import org.reactivestreams.Publisher;
-import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
 import reactor.core.Disposable;
+import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
@@ -474,28 +474,31 @@ void dispose() {
 
         void register(Client client) {
 
-            this.subscription = client.addNotificationListener(new Subscriber<NotificationResponse>() {
+            BaseSubscriber<NotificationResponse> subscriber = new BaseSubscriber<NotificationResponse>() {
 
                 @Override
-                public void onSubscribe(Subscription subscription) {
+                protected void hookOnSubscribe(Subscription subscription) {
                     subscription.request(Long.MAX_VALUE);
                 }
 
                 @Override
-                public void onNext(NotificationResponse notificationResponse) {
+                public void hookOnNext(NotificationResponse notificationResponse) {
                     NotificationAdapter.this.sink.emitNext(new NotificationResponseWrapper(notificationResponse), Sinks.EmitFailureHandler.FAIL_FAST);
                 }
 
                 @Override
-                public void onError(Throwable throwable) {
+                public void hookOnError(Throwable throwable) {
                     NotificationAdapter.this.sink.emitError(throwable, Sinks.EmitFailureHandler.FAIL_FAST);
                 }
 
                 @Override
-                public void onComplete() {
+                public void hookOnComplete() {
                     NotificationAdapter.this.sink.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
                 }
-            });
+            };
+
+            this.subscription = subscriber;
+            client.addNotificationListener(subscriber);
         }
 
         Flux<Notification> getEvents() {

src/main/java/io/r2dbc/postgresql/client/Client.java

@@ -57,7 +57,7 @@ public interface Client {
      * @throws IllegalArgumentException if {@code consumer} is {@code null}
      * @since 0.8.1
      */
-    Disposable addNotificationListener(Subscriber<NotificationResponse> consumer);
+    void addNotificationListener(Subscriber<NotificationResponse> consumer);
 
     /**
      * Release any resources held by the {@link Client}.

src/main/java/io/r2dbc/postgresql/client/ReactorNettyClient.java

@@ -137,8 +137,8 @@ private ReactorNettyClient(Connection connection, ConnectionSettings settings) {
         Assert.requireNonNull(connection, "Connection must not be null");
         this.settings = Assert.requireNonNull(settings, "ConnectionSettings must not be null");
 
-        connection.addHandler(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE - 5, 1, 4, -4, 0));
-        connection.addHandler(new EnsureSubscribersCompleteChannelHandler(this.requestSink));
+        connection.addHandlerFirst(new EnsureSubscribersCompleteChannelHandler(this.requestSink));
+        connection.addHandlerLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE - 5, 1, 4, -4, 0));
         this.connection = connection;
         this.byteBufAllocator = connection.outbound().alloc();
         this.context = new ConnectionContext().withChannelId(connection.channel().toString());
@@ -422,8 +422,8 @@ public Disposable addNotificationListener(Consumer<NotificationResponse> consume
     }
 
     @Override
-    public Disposable addNotificationListener(Subscriber<NotificationResponse> consumer) {
-        return this.notificationProcessor.asFlux().subscribe(consumer::onNext, consumer::onError, consumer::onComplete, consumer::onSubscribe);
+    public void addNotificationListener(Subscriber<NotificationResponse> consumer) {
+        this.notificationProcessor.asFlux().subscribe(consumer);
     }
 
     @Override
@@ -816,7 +816,7 @@ public void onComplete() {
         @Override
         public Context currentContext() {
             Conversation receiver = this.conversations.peek();
-            return receiver != null ? receiver.sink.currentContext() : Context.empty();
+            return receiver != null ? Context.of(receiver.sink.currentContext()) : Context.empty();
         }
 
         private void tryDrainLoop() {

src/test/java/io/r2dbc/postgresql/PostgresqlConnectionFactoryUnitTests.java

@@ -66,8 +66,10 @@ void createAuthenticationMD5Password() {
         // @formatter:off
         Client client = TestClient.builder()
             .window()
-                .expectRequest(new StartupMessage("test-application-name", "test-database", "test-username", Collections.emptyMap())).thenRespond(new AuthenticationMD5Password(TEST.buffer(4).writeInt(100)))
-                .expectRequest(new PasswordMessage("md55e9836cdb369d50e3bc7d127e88b4804")).thenRespond(AuthenticationOk.INSTANCE)
+                .expectRequest(new StartupMessage("test-application-name", "test-database", "test-username", Collections.emptyMap()))
+                .thenRespond(new AuthenticationMD5Password(TEST.buffer(4).writeInt(100)))
+                .expectRequest(new PasswordMessage("md55e9836cdb369d50e3bc7d127e88b4804"))
+                .thenRespond(AuthenticationOk.INSTANCE)
                 .done()
             .build();
         // @formatter:on

src/test/java/io/r2dbc/postgresql/client/TestClient.java

@@ -26,8 +26,8 @@
 import org.reactivestreams.Subscriber;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
 import reactor.util.annotation.Nullable;
 
 import java.util.ArrayList;
@@ -42,7 +42,6 @@
 /**
  * Test {@link Client} implementation that allows specification of expectations and assertions.
  */
-@SuppressWarnings("deprecation")
 public final class TestClient implements Client {
 
     public static final TestClient NO_OP = new TestClient(false, true, null, null, Flux.empty(), IDLE, new Version("9.4"));
@@ -51,15 +50,13 @@ public final class TestClient implements Client {
 
     private final boolean connected;
 
-    private final reactor.core.publisher.EmitterProcessor<NotificationResponse> notificationProcessor = reactor.core.publisher.EmitterProcessor.create(false);
+    private final Sinks.Many<NotificationResponse> notificationProcessor = Sinks.many().multicast().onBackpressureBuffer();
 
     private final Integer processId;
 
-    private final reactor.core.publisher.EmitterProcessor<FrontendMessage> requestProcessor = reactor.core.publisher.EmitterProcessor.create(false);
+    private final Sinks.Many<FrontendMessage> requestProcessor = Sinks.many().multicast().onBackpressureBuffer();
 
-    private final FluxSink<FrontendMessage> requests = this.requestProcessor.sink();
-
-    private final reactor.core.publisher.EmitterProcessor<Flux<BackendMessage>> responseProcessor = reactor.core.publisher.EmitterProcessor.create(false);
+    private final Sinks.Many<Flux<BackendMessage>> responseProcessor = Sinks.many().replay().all();
 
     private final Integer secretKey;
 
@@ -75,14 +72,12 @@ private TestClient(boolean expectClose, boolean connected, @Nullable Integer pro
         this.transactionStatus = Assert.requireNonNull(transactionStatus, "transactionStatus must not be null");
         this.version = version;
 
-        FluxSink<Flux<BackendMessage>> responses = this.responseProcessor.sink();
-
         Assert.requireNonNull(windows, "windows must not be null")
             .map(window -> window.exchanges)
             .map(exchanges -> exchanges
                 .concatMap(exchange ->
 
-                    this.requestProcessor.zipWith(exchange.requests)
+                    this.requestProcessor.asFlux().zipWith(exchange.requests)
                         .handle((tuple, sink) -> {
                             FrontendMessage actual = tuple.getT1();
                             FrontendMessage expected = tuple.getT2();
@@ -92,7 +87,7 @@ private TestClient(boolean expectClose, boolean connected, @Nullable Integer pro
                             }
                         })
                         .thenMany(exchange.responses)))
-            .subscribe(responses::next, responses::error, responses::complete);
+            .subscribe(this.responseProcessor::tryEmitNext, this.responseProcessor::tryEmitError, this.responseProcessor::tryEmitComplete);
     }
 
     public static Builder builder() {
@@ -108,20 +103,20 @@ public Mono<Void> close() {
     public Flux<BackendMessage> exchange(Publisher<FrontendMessage> requests) {
         Assert.requireNonNull(requests, "requests must not be null");
 
-        return this.responseProcessor
+        return this.responseProcessor.asFlux()
             .doOnSubscribe(s ->
                 Flux.from(requests)
-                    .subscribe(this.requests::next, this.requests::error))
+                    .subscribe(this.requestProcessor::tryEmitNext, this.requestProcessor::tryEmitError))
             .next()
             .flatMapMany(Function.identity());
     }
 
     @Override
     public Flux<BackendMessage> exchange(Predicate<BackendMessage> takeUntil, Publisher<FrontendMessage> requests) {
-        return this.responseProcessor
+        return this.responseProcessor.asFlux()
             .doOnSubscribe(s ->
                 Flux.from(requests)
-                    .subscribe(this.requests::next, this.requests::error))
+                    .subscribe(this.requestProcessor::tryEmitNext, this.requestProcessor::tryEmitError))
             .next()
             .flatMapMany(Function.identity())
             .takeWhile(takeUntil.negate());
@@ -169,21 +164,21 @@ public Mono<Void> cancelRequest() {
 
     @Override
     public void send(FrontendMessage message) {
-        this.requests.next(message);
+        this.requestProcessor.tryEmitNext(message);
     }
 
     @Override
     public Disposable addNotificationListener(Consumer<NotificationResponse> consumer) {
-        return this.notificationProcessor.subscribe(consumer);
+        return this.notificationProcessor.asFlux().subscribe(consumer);
     }
 
     @Override
-    public Disposable addNotificationListener(Subscriber<NotificationResponse> consumer) {
-        return this.notificationProcessor.subscribe(consumer::onNext, consumer::onError, consumer::onComplete, consumer::onSubscribe);
+    public void addNotificationListener(Subscriber<NotificationResponse> consumer) {
+        this.notificationProcessor.asFlux().subscribe(consumer);
     }
 
     public void notify(NotificationResponse notification) {
-        this.notificationProcessor.onNext(notification);
+        this.notificationProcessor.tryEmitNext(notification);
     }
 
     public static final class Builder {