Introduce new reactive session with updated API

driver/clirr-ignored-differences.xml

@@ -280,4 +280,28 @@
         <method>org.reactivestreams.Publisher isOpen()</method>
     </difference>
 
+    <difference>
+        <className>org/neo4j/driver/Driver</className>
+        <differenceType>7012</differenceType>
+        <method>org.neo4j.driver.reactive.ReactiveSession reactiveSession()</method>
+    </difference>
+
+    <difference>
+        <className>org/neo4j/driver/Driver</className>
+        <differenceType>7012</differenceType>
+        <method>org.neo4j.driver.reactive.ReactiveSession reactiveSession(org.neo4j.driver.SessionConfig)</method>
+    </difference>
+
+    <difference>
+        <className>org/neo4j/driver/reactive/RxQueryRunner</className>
+        <differenceType>7012</differenceType>
+        <method>org.neo4j.driver.Value parameters(org.neo4j.driver.Record)</method>
+    </difference>
+
+    <difference>
+        <className>org/neo4j/driver/reactive/RxQueryRunner</className>
+        <differenceType>7012</differenceType>
+        <method>org.neo4j.driver.Value parameters(java.util.Map)</method>
+    </difference>
+
 </differences>

driver/src/main/java/org/neo4j/driver/Driver.java

@@ -22,6 +22,7 @@
 
 import org.neo4j.driver.async.AsyncSession;
 import org.neo4j.driver.exceptions.ClientException;
+import org.neo4j.driver.reactive.ReactiveSession;
 import org.neo4j.driver.reactive.RxSession;
 import org.neo4j.driver.types.TypeSystem;
 import org.neo4j.driver.util.Experimental;
@@ -90,27 +91,57 @@ public interface Driver extends AutoCloseable
     Session session( SessionConfig sessionConfig );
 
     /**
-     * Create a new general purpose {@link RxSession} with default {@link SessionConfig session configuration}.
-     * The {@link RxSession} provides a reactive way to run queries and process results.
+     * Create a new general purpose {@link RxSession} with default {@link SessionConfig session configuration}. The {@link RxSession} provides a reactive way to
+     * run queries and process results.
      * <p>
      * Alias to {@link #rxSession(SessionConfig)}}.
      *
      * @return a new {@link RxSession} object.
+     * @deprecated superseded by {@link #reactiveSession()}.
      */
-    RxSession rxSession();
+    @Deprecated
+    default RxSession rxSession()
+    {
+        return rxSession( SessionConfig.defaultConfig() );
+    }
 
     /**
-     * Create a new {@link RxSession} with a specified {@link SessionConfig session configuration}.
-     * Use {@link SessionConfig#forDatabase(String)} to obtain a general purpose session configuration for the specified database.
-     * The {@link RxSession} provides a reactive way to run queries and process results.
+     * Create a new {@link RxSession} with a specified {@link SessionConfig session configuration}. Use {@link SessionConfig#forDatabase(String)} to obtain a
+     * general purpose session configuration for the specified database. The {@link RxSession} provides a reactive way to run queries and process results.
+     *
      * @param sessionConfig used to customize the session.
      * @return a new {@link RxSession} object.
+     * @deprecated superseded by {@link #reactiveSession(SessionConfig)}.
      */
+    @Deprecated
     RxSession rxSession( SessionConfig sessionConfig );
 
     /**
-     * Create a new general purpose {@link AsyncSession} with default {@link SessionConfig session configuration}.
-     * The {@link AsyncSession} provides an asynchronous way to run queries and process results.
+     * Create a new general purpose {@link ReactiveSession} with default {@link SessionConfig session configuration}. The {@link ReactiveSession} provides a
+     * reactive way to run queries and process results.
+     * <p>
+     * Alias to {@link #rxSession(SessionConfig)}}.
+     *
+     * @return a new {@link ReactiveSession} object.
+     */
+    default ReactiveSession reactiveSession()
+    {
+        return reactiveSession( SessionConfig.defaultConfig() );
+    }
+
+    /**
+     * Create a new {@link ReactiveSession} with a specified {@link SessionConfig session configuration}. Use {@link SessionConfig#forDatabase(String)} to
+     * obtain a general purpose session configuration for the specified database. The {@link ReactiveSession} provides a reactive way to run queries and process
+     * results.
+     *
+     * @param sessionConfig used to customize the session.
+     * @return a new {@link ReactiveSession} object.
+     */
+    ReactiveSession reactiveSession( SessionConfig sessionConfig );
+
+    /**
+     * Create a new general purpose {@link AsyncSession} with default {@link SessionConfig session configuration}. The {@link AsyncSession} provides an
+     * asynchronous way to run queries and process results.
      * <p>
      * Alias to {@link #asyncSession(SessionConfig)}}.
      *

driver/src/main/java/org/neo4j/driver/internal/InternalDriver.java

@@ -25,17 +25,19 @@
 import org.neo4j.driver.Logger;
 import org.neo4j.driver.Logging;
 import org.neo4j.driver.Metrics;
-import org.neo4j.driver.internal.metrics.MetricsProvider;
 import org.neo4j.driver.Session;
 import org.neo4j.driver.SessionConfig;
 import org.neo4j.driver.async.AsyncSession;
 import org.neo4j.driver.internal.async.InternalAsyncSession;
 import org.neo4j.driver.internal.async.NetworkSession;
 import org.neo4j.driver.internal.metrics.DevNullMetricsProvider;
+import org.neo4j.driver.internal.metrics.MetricsProvider;
+import org.neo4j.driver.internal.reactive.InternalReactiveSession;
 import org.neo4j.driver.internal.reactive.InternalRxSession;
 import org.neo4j.driver.internal.security.SecurityPlan;
 import org.neo4j.driver.internal.types.InternalTypeSystem;
 import org.neo4j.driver.internal.util.Futures;
+import org.neo4j.driver.reactive.ReactiveSession;
 import org.neo4j.driver.reactive.RxSession;
 import org.neo4j.driver.types.TypeSystem;
 
@@ -71,15 +73,15 @@ public Session session( SessionConfig sessionConfig )
     }
 
     @Override
-    public RxSession rxSession()
+    public RxSession rxSession( SessionConfig sessionConfig )
     {
-        return new InternalRxSession( newSession( SessionConfig.defaultConfig() ) );
+        return new InternalRxSession( newSession( sessionConfig ) );
     }
 
     @Override
-    public RxSession rxSession( SessionConfig sessionConfig )
+    public ReactiveSession reactiveSession( SessionConfig sessionConfig )
     {
-        return new InternalRxSession( newSession( sessionConfig ) );
+        return new InternalReactiveSession( newSession( sessionConfig ) );
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/cursor/RxResultCursor.java

@@ -37,4 +37,6 @@ public interface RxResultCursor extends Subscription, FailableCursor
     CompletionStage<ResultSummary> summaryAsync();
 
     boolean isDone();
+
+    Throwable getRunError();
 }

driver/src/main/java/org/neo4j/driver/internal/cursor/RxResultCursorImpl.java

@@ -42,6 +42,7 @@ public class RxResultCursorImpl implements RxResultCursor
     private final RunResponseHandler runHandler;
     private final PullResponseHandler pullHandler;
     private final Throwable runResponseError;
+    private boolean runErrorSurfaced;
     private final CompletableFuture<ResultSummary> summaryFuture = new CompletableFuture<>();
     private boolean summaryFutureExposed;
     private boolean resultConsumed;
@@ -108,7 +109,7 @@ public CompletionStage<Throwable> discardAllFailureAsync()
     {
         // calling this method will enforce discarding record stream and finish running cypher query
         return summaryStage().thenApply( summary -> (Throwable) null )
-                             .exceptionally( throwable -> summaryFutureExposed ? null : throwable );
+                             .exceptionally( throwable -> runErrorSurfaced || summaryFutureExposed ? null : throwable );
     }
 
     @Override
@@ -136,6 +137,13 @@ public boolean isDone()
         return summaryFuture.isDone();
     }
 
+    @Override
+    public Throwable getRunError()
+    {
+        runErrorSurfaced = true;
+        return runResponseError;
+    }
+
     public CompletionStage<ResultSummary> summaryStage()
     {
         if ( !isDone() && !resultConsumed ) // the summary is called before record streaming

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

@@ -0,0 +1,136 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.driver.internal.reactive;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+
+import org.neo4j.driver.AccessMode;
+import org.neo4j.driver.Bookmark;
+import org.neo4j.driver.TransactionConfig;
+import org.neo4j.driver.exceptions.TransactionNestingException;
+import org.neo4j.driver.internal.async.NetworkSession;
+import org.neo4j.driver.internal.async.UnmanagedTransaction;
+import org.neo4j.driver.internal.util.Futures;
+
+import static org.neo4j.driver.internal.reactive.RxUtils.createEmptyPublisher;
+import static org.neo4j.driver.internal.reactive.RxUtils.createSingleItemPublisher;
+
+abstract class AbstractReactiveSession<S>
+{
+    protected final NetworkSession session;
+
+    public AbstractReactiveSession( NetworkSession session )
+    {
+        // RxSession accept a network session as input.
+        // The network session different from async session that it provides ways to both run for Rx and Async
+        // Note: Blocking result could just build on top of async result. However, Rx result cannot just build on top of async result.
+        this.session = session;
+    }
+
+    abstract S createTransaction( UnmanagedTransaction unmanagedTransaction );
+
+    abstract Publisher<Void> closeTransaction( S transaction, boolean commit );
+
+    public Publisher<S> beginTransaction( TransactionConfig config )
+    {
+        return createSingleItemPublisher(
+                () ->
+                {
+                    CompletableFuture<S> txFuture = new CompletableFuture<>();
+                    session.beginTransactionAsync( config ).whenComplete(
+                            ( tx, completionError ) ->
+                            {
+                                if ( tx != null )
+                                {
+                                    txFuture.complete( createTransaction( tx ) );
+                                }
+                                else
+                                {
+                                    releaseConnectionBeforeReturning( txFuture, completionError );
+                                }
+                            } );
+                    return txFuture;
+                }, () -> new IllegalStateException( "Unexpected condition, begin transaction call has completed successfully with transaction being null" ) );
+    }
+
+    Publisher<S> beginTransaction( AccessMode mode, TransactionConfig config )
+    {
+        return createSingleItemPublisher(
+                () ->
+                {
+                    CompletableFuture<S> txFuture = new CompletableFuture<>();
+                    session.beginTransactionAsync( mode, config ).whenComplete(
+                            ( tx, completionError ) ->
+                            {
+                                if ( tx != null )
+                                {
+                                    txFuture.complete( createTransaction( tx ) );
+                                }
+                                else
+                                {
+                                    releaseConnectionBeforeReturning( txFuture, completionError );
+                                }
+                            } );
+                    return txFuture;
+                }, () -> new IllegalStateException( "Unexpected condition, begin transaction call has completed successfully with transaction being null" ) );
+    }
+
+    <T> Publisher<T> runTransaction( AccessMode mode, Function<S,? extends Publisher<T>> work, TransactionConfig config )
+    {
+        Flux<T> repeatableWork = Flux.usingWhen( beginTransaction( mode, config ),
+                                                 work,
+                                                 tx -> closeTransaction( tx, true ),
+                                                 ( tx, error ) -> closeTransaction( tx, false ),
+                                                 ( tx ) -> closeTransaction( tx, false ) );
+        return session.retryLogic().retryRx( repeatableWork );
+    }
+
+    private <T> void releaseConnectionBeforeReturning( CompletableFuture<T> returnFuture, Throwable completionError )
+    {
+        // We failed to create a result cursor, so we cannot rely on result cursor to clean-up resources.
+        // Therefore, we will first release the connection that might have been created in the session and then notify the error.
+        // The logic here shall be the same as `SessionPullResponseHandler#afterFailure`.
+        // The reason we need to release connection in session is that we made `rxSession.close()` optional;
+        // Otherwise, session.close shall handle everything for us.
+        Throwable error = Futures.completionExceptionCause( completionError );
+        if ( error instanceof TransactionNestingException )
+        {
+            returnFuture.completeExceptionally( error );
+        }
+        else
+        {
+            session.releaseConnectionAsync().whenComplete( ( ignored, closeError ) ->
+                                                                   returnFuture.completeExceptionally( Futures.combineErrors( error, closeError ) ) );
+        }
+    }
+
+    public Bookmark lastBookmark()
+    {
+        return session.lastBookmark();
+    }
+
+    public <T> Publisher<T> close()
+    {
+        return createEmptyPublisher( session::closeAsync );
+    }
+}

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

@@ -0,0 +1,61 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.driver.internal.reactive;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
+
+import org.neo4j.driver.internal.async.UnmanagedTransaction;
+
+import static org.neo4j.driver.internal.reactive.RxUtils.createEmptyPublisher;
+
+abstract class AbstractReactiveTransaction
+{
+    protected final UnmanagedTransaction tx;
+
+    protected AbstractReactiveTransaction( UnmanagedTransaction tx )
+    {
+        this.tx = tx;
+    }
+
+    public <T> Publisher<T> commit()
+    {
+        return createEmptyPublisher( tx::commitAsync );
+    }
+
+    public <T> Publisher<T> rollback()
+    {
+        return createEmptyPublisher( tx::rollbackAsync );
+    }
+
+    public Publisher<Void> close()
+    {
+        return close( false );
+    }
+
+    public Publisher<Boolean> isOpen()
+    {
+        return Mono.just( tx.isOpen() );
+    }
+
+    Publisher<Void> close( boolean commit )
+    {
+        return createEmptyPublisher( () -> tx.closeAsync( commit ) );
+    }
+}