Add support for cancellation on reactive session run

This update brings support for cancelling subscription on reactive session `run` methods, like:
- `org.neo4j.driver.reactive.ReactiveSession.run(..)`
- `org.neo4j.driver.reactivestreams.ReactiveSession.run(..)`

If subscription is cancelled before an instance of the `ReactiveResult` is published, the driver will rollback the initiated query by sending the Bolt `RESET` message and will release the network connection back to its connection pool. This will be done in the background.

Once the `ReactiveResult` is published, the cancellation with the rollback is no longer possible.

In addition, this update makes the driver wait until subscriber subscribes to the publisher before it start network communication. Previously, the driver would do that immediate upon calling the `run` method.

Author: injectives

driver/src/main/java/org/neo4j/driver/internal/reactive/AbstractReactiveSession.java

@@ -21,7 +21,6 @@
 import static org.neo4j.driver.internal.reactive.RxUtils.createEmptyPublisher;
 import static org.neo4j.driver.internal.reactive.RxUtils.createSingleItemPublisher;
 
-import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
@@ -151,65 +150,27 @@ public Set<Bookmark> lastBookmarks() {
     protected <T> Publisher<T> run(Query query, TransactionConfig config, Function<RxResultCursor, T> cursorToResult) {
         var cursorPublishFuture = new CompletableFuture<RxResultCursor>();
         var cursorReference = new AtomicReference<RxResultCursor>();
-        return Mono.<T>create(sink -> {
-                    var state = new SinkState<T>();
-                    sink.onRequest(ignored -> {
-                        CompletionStage<T> stage;
-                        synchronized (state) {
-                            if (state.isCancelled()) {
-                                return;
-                            }
-                            if (state.getStage() != null) {
-                                return;
-                            }
-                            stage = runAsStage(query, config, cursorPublishFuture)
-                                    .thenApply(cursor -> {
-                                        cursorReference.set(cursor);
-                                        return cursor;
-                                    })
-                                    .thenApply(cursorToResult);
-                            state.setStage(stage);
-                        }
-                        stage.whenComplete((item, completionError) -> {
-                            if (completionError == null) {
-                                if (item != null) {
-                                    sink.success(item);
-                                } else {
-                                    sink.error(
-                                            new IllegalStateException(
-                                                    "Unexpected condition, run call has completed successfully with result being null"));
-                                }
-                            } else {
-                                var error = Optional.ofNullable(Futures.completionExceptionCause(completionError))
-                                        .orElse(completionError);
-                                sink.error(error);
-                            }
-                        });
-                    });
-                    sink.onCancel(() -> {
-                        CompletionStage<T> stage;
-                        synchronized (state) {
-                            if (state.isCancelled()) {
-                                return;
+
+        return createSingleItemPublisher(
+                        () -> runAsStage(query, config, cursorPublishFuture)
+                                .thenApply(cursor -> {
+                                    cursorReference.set(cursor);
+                                    return cursor;
+                                })
+                                .thenApply(cursorToResult),
+                        () -> new IllegalStateException(
+                                "Unexpected condition, run call has completed successfully with result being null"),
+                        value -> {
+                            if (value != null) {
+                                cursorReference.get().rollback().whenComplete((unused, throwable) -> {
+                                    if (throwable != null) {
+                                        cursorPublishFuture.completeExceptionally(throwable);
+                                    } else {
+                                        cursorPublishFuture.complete(null);
+                                    }
+                                });
                             }
-                            state.setCancelled(true);
-                            stage = state.getStage();
-                        }
-                        if (stage != null) {
-                            stage.whenComplete((value, ignored) -> {
-                                if (value != null) {
-                                    cursorReference.get().rollback().whenComplete((unused, throwable) -> {
-                                        if (throwable != null) {
-                                            cursorPublishFuture.completeExceptionally(throwable);
-                                        } else {
-                                            cursorPublishFuture.complete(null);
-                                        }
-                                    });
-                                }
-                            });
-                        }
-                    });
-                })
+                        })
                 .doOnNext(value -> cursorPublishFuture.complete(cursorReference.get()))
                 .doOnError(cursorPublishFuture::completeExceptionally);
     }
@@ -256,25 +217,4 @@ private <T> CompletionStage<T> releaseConnectionAndRethrow(Throwable throwable)
     protected <T> Publisher<T> doClose() {
         return createEmptyPublisher(session::closeAsync);
     }
-
-    private static class SinkState<T> {
-        private CompletionStage<T> stage;
-        private boolean cancelled;
-
-        public CompletionStage<T> getStage() {
-            return stage;
-        }
-
-        public void setStage(CompletionStage<T> stage) {
-            this.stage = stage;
-        }
-
-        public boolean isCancelled() {
-            return cancelled;
-        }
-
-        public void setCancelled(boolean cancelled) {
-            this.cancelled = cancelled;
-        }
-    }
 }

driver/src/main/java/org/neo4j/driver/internal/reactive/RxUtils.java

@@ -56,7 +56,7 @@ public static <T> Publisher<T> createEmptyPublisher(Supplier<CompletionStage<Voi
      * @param <T>                         the type of the item to publish.
      * @return A publisher that succeeds exactly one item or fails with an error.
      */
-    public static <T> Publisher<T> createSingleItemPublisher(
+    public static <T> Mono<T> createSingleItemPublisher(
             Supplier<CompletionStage<T>> supplier,
             Supplier<Throwable> nullResultThrowableSupplier,
             Consumer<T> cancellationHandler) {