Fix request queue delaying (#119)

Motivation:
The request queue could contain cancelled tasks, which caused delays and
stalls.

Modification:
Implemented Filtered out cancelled tasks from the queue.

Result:
The queue now functions better.
Resolves issue #114

Author: jchrys

src/main/java/io/asyncer/r2dbc/mysql/client/RequestQueue.java

@@ -65,22 +65,29 @@ final class RequestQueue extends ActiveStatus implements Runnable {
      */
     @Override
     public void run() {
-        RequestTask<?> task = queue.poll();
-
-        if (task == null) {
-            // Queue was empty, set it to idle if it is not disposed.
-            STATUS_UPDATER.compareAndSet(this, ACTIVE, IDLE);
-        } else {
-            int status = this.status;
-
-            if (status == DISPOSE) {
+        for (;;) {
+            RequestTask<?> task = queue.poll();
+            final int status = this.status;
+
+            if (task == null) {
+                // Queue was empty, set it to idle if it is not disposed.
+                if (status != ACTIVE || STATUS_UPDATER.compareAndSet(this, ACTIVE, IDLE) && queue.isEmpty()) {
+                    return;
+                }
+            } else if (status == DISPOSE) {
                 // Cancel and no need clear queue because it should be cleared by other one.
                 task.cancel(requireDisposed());
+                return;
             } else {
                 task.run();
+                // The execution of a canceled task would result in a stall of the request queue.
+                // refer: https://github.com/asyncer-io/r2dbc-mysql/issues/114
+                if (!task.isCancelled()) {
+                    return;
             }
         }
     }
+}
 
     /**
      * Submit an exchange task. If the queue is inactive, it will execute directly instead of queuing.
@@ -90,13 +97,6 @@ public void run() {
      * @param <T> the type argument of {@link RequestTask}.
      */
     <T> void submit(RequestTask<T> task) {
-        if (STATUS_UPDATER.compareAndSet(this, IDLE, ACTIVE)) {
-            // Fast path for general way.
-            task.run();
-            return;
-        }
-
-        // Check dispose after fast path failed.
         int status = this.status;
 
         if (status == DISPOSE) {

src/main/java/io/asyncer/r2dbc/mysql/client/RequestTask.java

@@ -38,6 +38,8 @@ final class RequestTask<T> {
 
     private final T supplier;
 
+    private volatile boolean isCancelled;
+
     private RequestTask(@Nullable Disposable disposable, MonoSink<T> sink, T supplier) {
         this.disposable = disposable;
         this.sink = sink;
@@ -54,26 +56,43 @@ void run() {
      * @param e cancelled by which error
      */
     void cancel(Throwable e) {
+        cancel0();
+        sink.error(e);
+    }
+
+    boolean isCancelled() {
+        return isCancelled;
+    }
+
+    private void cancel0() {
         if (disposable != null) {
             disposable.dispose();
         }
-        sink.error(e);
+        isCancelled = true;
     }
 
     static <T> RequestTask<T> wrap(ClientMessage message, MonoSink<T> sink, T supplier) {
+        final RequestTask<T> task;
         if (message instanceof Disposable) {
-            return new RequestTask<>((Disposable) message, sink, supplier);
-        }
+            task = new RequestTask<>((Disposable) message, sink, supplier);
+        } else {
+            task = new RequestTask<>(null, sink, supplier);
 
-        return new RequestTask<>(null, sink, supplier);
+        }
+        sink.onCancel(() -> task.cancel0());
+        return task;
     }
 
     static <T> RequestTask<T> wrap(Flux<? extends ClientMessage> messages, MonoSink<T> sink, T supplier) {
-        return new RequestTask<>(new DisposableFlux(messages), sink, supplier);
+        final RequestTask<T> task =  new RequestTask<>(new DisposableFlux(messages), sink, supplier);
+        sink.onCancel(() -> task.cancel0());
+        return task;
     }
 
     static <T> RequestTask<T> wrap(MonoSink<T> sink, T supplier) {
-        return new RequestTask<>(null, sink, supplier);
+        final RequestTask<T> task = new RequestTask<>(null, sink, supplier);
+        sink.onCancel(() -> task.cancel0());
+        return task;
     }
 
     private static final class DisposableFlux implements Disposable {

src/test/java/io/asyncer/r2dbc/mysql/ConnectionIntegrationTest.java

@@ -233,6 +233,16 @@ void setTransactionIsolationLevel() {
                 .doOnNext(a -> a.isEqualTo(connection.getTransactionIsolationLevel()))));
     }
 
+    @Test
+    void errorPropagteRequestQueue() {
+        illegalArgument(connection -> Flux.merge(
+                                connection.createStatement("SELECT 'Result 1', SLEEP(1)").execute(),
+                                connection.createStatement("SELECT 'Result 2'").execute(),
+                                connection.createStatement("SELECT 'Result 3'").execute()
+                        ).flatMap(result -> result.map((row, meta) -> row.get(0, Integer.class)))
+        );
+    }
+
     @Test
     void batchCrud() {
         // TODO: spilt it to multiple test cases and move it to BatchIntegrationTest

src/test/java/io/asyncer/r2dbc/mysql/IntegrationTestSupport.java

@@ -50,6 +50,10 @@ void badGrammar(Function<? super MySqlConnection, Publisher<?>> runner) {
         process(runner).verifyError(R2dbcBadGrammarException.class);
     }
 
+    void illegalArgument(Function<? super MySqlConnection, Publisher<?>> runner) {
+        process(runner).expectError(IllegalArgumentException.class).verify(Duration.ofSeconds(3));
+    }
+
     Mono<MySqlConnection> create() {
         return connectionFactory.create();
     }