Terminate request queue on connection error first before terminating response handlers.

mp911de · mp911de · commit 28e24aaf6952 · 2023-11-20T13:49:33.000+01:00
Terminating the request queue will ensure that no new requests will be accepted and thus that we could miss a response handler to be terminated. [#245] Signed-off-by: Mark Paluch <mpaluch@vmware.com>
diff --git a/src/main/java/io/r2dbc/mssql/client/ReactorNettyClient.java b/src/main/java/io/r2dbc/mssql/client/ReactorNettyClient.java
@@ -36,12 +36,7 @@
 import io.r2dbc.mssql.message.header.PacketIdProvider;
 import io.r2dbc.mssql.message.tds.ProtocolException;
 import io.r2dbc.mssql.message.tds.Redirect;
-import io.r2dbc.mssql.message.token.AbstractDoneToken;
-import io.r2dbc.mssql.message.token.AbstractInfoToken;
-import io.r2dbc.mssql.message.token.Attention;
-import io.r2dbc.mssql.message.token.EnvChangeToken;
-import io.r2dbc.mssql.message.token.FeatureExtAckToken;
-import io.r2dbc.mssql.message.token.LoginAckToken;
+import io.r2dbc.mssql.message.token.*;
 import io.r2dbc.mssql.message.type.Collation;
 import io.r2dbc.mssql.util.Assert;
 import io.r2dbc.spi.R2dbcException;
@@ -50,11 +45,7 @@
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoSink;
-import reactor.core.publisher.Sinks;
-import reactor.core.publisher.SynchronousSink;
+import reactor.core.publisher.*;
 import reactor.netty.Connection;
 import reactor.netty.NettyOutbound;
 import reactor.netty.resources.ConnectionProvider;
@@ -367,7 +358,8 @@ private Object encodeForSend(ClientMessage message) {
     @SuppressWarnings("unchecked")
     private <T> Mono<T> resumeError(Throwable throwable) {
 
-        handleConnectionError(throwable);
+        logger.error(this.context.getMessage("Error: {}"), throwable.getMessage(), throwable);
+
         this.requestSink.emitComplete((signalType, emitResult) -> {
 
             if (emitResult.isFailure()) {
@@ -377,8 +369,7 @@ private <T> Mono<T> resumeError(Throwable throwable) {
             return false;
         });
 
-        logger.error(this.context.getMessage("Error: {}"), throwable.getMessage(), throwable);
-
+        handleConnectionError(throwable);
         return (Mono<T>) close();
     }
 
