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

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 <[email protected]>

Author: mp911de

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

@@ -36,25 +36,15 @@
 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;
 import io.r2dbc.spi.R2dbcNonTransientResourceException;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
-import reactor.core.publisher.EmitterProcessor;
-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;
@@ -363,7 +353,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()) {
@@ -373,8 +364,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();
     }