Added async transaction functions

Via `Session#readTransactionAsync()` and `#writeTransactionAsync()`.
Both accept a single parameter - async function that takes a
transaction and returns a `Response`. When function returns a failed
response or throws exception it will be retried. Retries only happen
on non-fatal & transient errors.

API is not fully complete yet - there is no way to create a `Result`
object so it'll be inconvenient to use. Helpers to create some sort
of "settable" response will come in subsequent commits.

Author: lutovich

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

@@ -19,6 +19,7 @@
 package org.neo4j.driver.internal;
 
 import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.EventLoopGroup;
 import io.netty.util.concurrent.EventExecutorGroup;
 
 import java.io.IOException;
@@ -75,15 +76,16 @@ public final Driver newInstance( URI uri, AuthToken authToken, RoutingSettings r
         ConnectionPool connectionPool = createConnectionPool( authToken, securityPlan, config );
 
         Bootstrap bootstrap = BootstrapFactory.newBootstrap();
-        RetryLogic retryLogic = createRetryLogic( retrySettings, bootstrap.config().group(), config.logging() );
+        EventLoopGroup eventLoopGroup = bootstrap.config().group();
+        RetryLogic retryLogic = createRetryLogic( retrySettings, eventLoopGroup, config.logging() );
 
         AsyncConnectionPool asyncConnectionPool = createAsyncConnectionPool( authToken, securityPlan, bootstrap,
                 config );
 
         try
         {
             return createDriver( uri, address, connectionPool, config, newRoutingSettings, securityPlan, retryLogic,
-                    asyncConnectionPool );
+                    asyncConnectionPool, eventLoopGroup );
         }
         catch ( Throwable driverError )
         {
@@ -119,16 +121,18 @@ private AsyncConnectionPool createAsyncConnectionPool( AuthToken authToken, Secu
 
     private Driver createDriver( URI uri, BoltServerAddress address, ConnectionPool connectionPool,
             Config config, RoutingSettings routingSettings, SecurityPlan securityPlan,
-            RetryLogic retryLogic, AsyncConnectionPool asyncConnectionPool )
+            RetryLogic retryLogic, AsyncConnectionPool asyncConnectionPool, EventExecutorGroup eventExecutorGroup )
     {
         String scheme = uri.getScheme().toLowerCase();
         switch ( scheme )
         {
         case BOLT_URI_SCHEME:
             assertNoRoutingContext( uri, routingSettings );
-            return createDirectDriver( address, connectionPool, config, securityPlan, retryLogic, asyncConnectionPool );
+            return createDirectDriver( address, connectionPool, config, securityPlan, retryLogic, asyncConnectionPool,
+                    eventExecutorGroup );
         case BOLT_ROUTING_URI_SCHEME:
-            return createRoutingDriver( address, connectionPool, config, routingSettings, securityPlan, retryLogic );
+            return createRoutingDriver( address, connectionPool, config, routingSettings, securityPlan, retryLogic,
+                    eventExecutorGroup );
         default:
             throw new ClientException( format( "Unsupported URI scheme: %s", scheme ) );
         }
@@ -140,11 +144,13 @@ private Driver createDriver( URI uri, BoltServerAddress address, ConnectionPool
      * <b>This method is protected only for testing</b>
      */
     protected Driver createDirectDriver( BoltServerAddress address, ConnectionPool connectionPool, Config config,
-            SecurityPlan securityPlan, RetryLogic retryLogic, AsyncConnectionPool asyncConnectionPool )
+            SecurityPlan securityPlan, RetryLogic retryLogic, AsyncConnectionPool asyncConnectionPool,
+            EventExecutorGroup eventExecutorGroup )
     {
         ConnectionProvider connectionProvider =
                 new DirectConnectionProvider( address, connectionPool, asyncConnectionPool );
-        SessionFactory sessionFactory = createSessionFactory( connectionProvider, retryLogic, config );
+        SessionFactory sessionFactory =
+                createSessionFactory( connectionProvider, retryLogic, eventExecutorGroup, config );
         return createDriver( config, securityPlan, sessionFactory );
     }
 
@@ -154,14 +160,16 @@ protected Driver createDirectDriver( BoltServerAddress address, ConnectionPool c
      * <b>This method is protected only for testing</b>
      */
     protected Driver createRoutingDriver( BoltServerAddress address, ConnectionPool connectionPool,
-            Config config, RoutingSettings routingSettings, SecurityPlan securityPlan, RetryLogic retryLogic )
+            Config config, RoutingSettings routingSettings, SecurityPlan securityPlan, RetryLogic retryLogic,
+            EventExecutorGroup eventExecutorGroup )
     {
         if ( !securityPlan.isRoutingCompatible() )
         {
             throw new IllegalArgumentException( "The chosen security plan is not compatible with a routing driver" );
         }
         ConnectionProvider connectionProvider = createLoadBalancer( address, connectionPool, config, routingSettings );
-        SessionFactory sessionFactory = createSessionFactory( connectionProvider, retryLogic, config );
+        SessionFactory sessionFactory =
+                createSessionFactory( connectionProvider, retryLogic, eventExecutorGroup, config );
         return createDriver( config, securityPlan, sessionFactory );
     }
 
@@ -242,10 +250,10 @@ protected Connector createConnector( final ConnectionSettings connectionSettings
      * <p>
      * <b>This method is protected only for testing</b>
      */
-    protected SessionFactory createSessionFactory( ConnectionProvider connectionProvider,
-            RetryLogic retryLogic, Config config )
+    protected SessionFactory createSessionFactory( ConnectionProvider connectionProvider, RetryLogic retryLogic,
+            EventExecutorGroup eventExecutorGroup, Config config )
     {
-        return new SessionFactoryImpl( connectionProvider, retryLogic, config );
+        return new SessionFactoryImpl( connectionProvider, retryLogic, eventExecutorGroup, config );
     }
 
     /**

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

@@ -18,6 +18,8 @@
  */
 package org.neo4j.driver.internal;
 
+import io.netty.util.concurrent.EventExecutorGroup;
+
 import org.neo4j.driver.internal.retry.RetryLogic;
 import org.neo4j.driver.internal.spi.ConnectionProvider;
 import org.neo4j.driver.v1.AccessMode;
@@ -30,9 +32,9 @@ class LeakLoggingNetworkSession extends NetworkSession
     private final String stackTrace;
 
     LeakLoggingNetworkSession( ConnectionProvider connectionProvider, AccessMode mode, RetryLogic retryLogic,
-            Logging logging )
+            EventExecutorGroup eventExecutorGroup, Logging logging )
     {
-        super( connectionProvider, mode, retryLogic, logging );
+        super( connectionProvider, mode, retryLogic, eventExecutorGroup, logging );
         this.stackTrace = captureStackTrace();
     }
 

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

@@ -18,7 +18,9 @@
  */
 package org.neo4j.driver.internal;
 
-import io.netty.util.concurrent.GlobalEventExecutor;
+import io.netty.util.concurrent.EventExecutorGroup;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.FutureListener;
 
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -40,6 +42,7 @@
 import org.neo4j.driver.v1.Logging;
 import org.neo4j.driver.v1.Record;
 import org.neo4j.driver.v1.Response;
+import org.neo4j.driver.v1.ResponseListener;
 import org.neo4j.driver.v1.Session;
 import org.neo4j.driver.v1.Statement;
 import org.neo4j.driver.v1.StatementResult;
@@ -61,6 +64,7 @@ public class NetworkSession implements Session, SessionResourcesHandler, ResultR
     private final ConnectionProvider connectionProvider;
     private final AccessMode mode;
     private final RetryLogic retryLogic;
+    private final EventExecutorGroup eventExecutorGroup;
     protected final Logger logger;
 
     private volatile Bookmark bookmark = Bookmark.empty();
@@ -73,11 +77,12 @@ public class NetworkSession implements Session, SessionResourcesHandler, ResultR
     private final AtomicBoolean isOpen = new AtomicBoolean( true );
 
     public NetworkSession( ConnectionProvider connectionProvider, AccessMode mode, RetryLogic retryLogic,
-            Logging logging )
+            EventExecutorGroup eventExecutorGroup, Logging logging )
     {
         this.connectionProvider = connectionProvider;
         this.mode = mode;
         this.retryLogic = retryLogic;
+        this.eventExecutorGroup = eventExecutorGroup;
         this.logger = new DelegatingLogger( logging.getLog( LOG_NAME ), String.valueOf( hashCode() ) );
     }
 
@@ -262,7 +267,7 @@ public InternalFuture<Void> apply( ExplicitTransaction tx )
         }
         else
         {
-            return new InternalPromise<Void>( GlobalEventExecutor.INSTANCE ).setSuccess( null );
+            return new InternalPromise<Void>( eventExecutorGroup ).setSuccess( null );
         }
     }
 
@@ -293,12 +298,24 @@ public <T> T readTransaction( TransactionWork<T> work )
         return transaction( AccessMode.READ, work );
     }
 
+    @Override
+    public <T> Response<T> readTransactionAsync( TransactionWork<Response<T>> work )
+    {
+        return transactionAsync( AccessMode.READ, work );
+    }
+
     @Override
     public <T> T writeTransaction( TransactionWork<T> work )
     {
         return transaction( AccessMode.WRITE, work );
     }
 
+    @Override
+    public <T> Response<T> writeTransactionAsync( TransactionWork<Response<T>> work )
+    {
+        return transactionAsync( AccessMode.WRITE, work );
+    }
+
     void setBookmark( Bookmark bookmark )
     {
         if ( bookmark != null && !bookmark.isEmpty() )
@@ -398,6 +415,113 @@ public T get()
         } );
     }
 
+    private <T> InternalFuture<T> transactionAsync( final AccessMode mode, final TransactionWork<Response<T>> work )
+    {
+        return retryLogic.retryAsync( new Supplier<InternalFuture<T>>()
+        {
+            @Override
+            public InternalFuture<T> get()
+            {
+                final InternalFuture<ExplicitTransaction> txFuture = beginTransactionAsync( mode );
+                final InternalPromise<T> resultPromise = new InternalPromise<>( txFuture.eventExecutor() );
+
+                txFuture.addListener( new FutureListener<ExplicitTransaction>()
+                {
+                    @Override
+                    public void operationComplete( Future<ExplicitTransaction> future ) throws Exception
+                    {
+                        if ( future.isCancelled() )
+                        {
+                            resultPromise.cancel( true );
+                        }
+                        else if ( future.isSuccess() )
+                        {
+                            executeWork( resultPromise, future.getNow(), work );
+                        }
+                        else
+                        {
+                            resultPromise.setFailure( future.cause() );
+                        }
+                    }
+                } );
+
+                return resultPromise;
+            }
+        } );
+    }
+
+    private <T> void executeWork( final InternalPromise<T> resultPromise, final ExplicitTransaction tx,
+            TransactionWork<Response<T>> work )
+    {
+        Response<T> workResponse = work.execute( tx );
+        workResponse.addListener( new ResponseListener<T>()
+        {
+            @Override
+            public void operationCompleted( T result, Throwable error )
+            {
+                if ( error != null )
+                {
+                    rollbackTxAfterFailedTransactionWork( tx, resultPromise, error );
+                }
+                else
+                {
+                    commitTxAfterSucceededTransactionWork( tx, resultPromise, result );
+                }
+            }
+        } );
+    }
+
+    private <T> void rollbackTxAfterFailedTransactionWork( ExplicitTransaction tx,
+            final InternalPromise<T> resultPromise, final Throwable error )
+    {
+        if ( tx.isOpen() )
+        {
+            tx.rollbackAsync().addListener( new ResponseListener<Void>()
+            {
+                @Override
+                public void operationCompleted( Void ignore, Throwable rollbackError )
+                {
+                    if ( rollbackError != null )
+                    {
+                        error.addSuppressed( rollbackError );
+                    }
+                    resultPromise.setFailure( error );
+                }
+            } );
+        }
+        else
+        {
+            resultPromise.setFailure( error );
+        }
+    }
+
+    private <T> void commitTxAfterSucceededTransactionWork( ExplicitTransaction tx,
+            final InternalPromise<T> resultPromise, final T result )
+    {
+        if ( tx.isOpen() )
+        {
+            tx.commitAsync().addListener( new ResponseListener<Void>()
+            {
+                @Override
+                public void operationCompleted( Void ignore, Throwable commitError )
+                {
+                    if ( commitError != null )
+                    {
+                        resultPromise.setFailure( commitError );
+                    }
+                    else
+                    {
+                        resultPromise.setSuccess( result );
+                    }
+                }
+            } );
+        }
+        else
+        {
+            resultPromise.setSuccess( result );
+        }
+    }
+
     private synchronized Transaction beginTransaction( AccessMode mode )
     {
         ensureSessionIsOpen();
@@ -431,7 +555,6 @@ public InternalFuture<ExplicitTransaction> apply( AsyncConnection connection )
                     }
                 } );
 
-        //noinspection unchecked
         return currentAsyncTransactionFuture;
     }
 

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

@@ -18,6 +18,8 @@
  */
 package org.neo4j.driver.internal;
 
+import io.netty.util.concurrent.EventExecutorGroup;
+
 import org.neo4j.driver.internal.retry.RetryLogic;
 import org.neo4j.driver.internal.spi.ConnectionProvider;
 import org.neo4j.driver.v1.AccessMode;
@@ -29,12 +31,15 @@ public class SessionFactoryImpl implements SessionFactory
 {
     private final ConnectionProvider connectionProvider;
     private final RetryLogic retryLogic;
+    private final EventExecutorGroup eventExecutorGroup;
     private final Logging logging;
     private final boolean leakedSessionsLoggingEnabled;
 
-    SessionFactoryImpl( ConnectionProvider connectionProvider, RetryLogic retryLogic, Config config )
+    SessionFactoryImpl( ConnectionProvider connectionProvider, RetryLogic retryLogic,
+            EventExecutorGroup eventExecutorGroup, Config config )
     {
         this.connectionProvider = connectionProvider;
+        this.eventExecutorGroup = eventExecutorGroup;
         this.leakedSessionsLoggingEnabled = config.logLeakedSessions();
         this.retryLogic = retryLogic;
         this.logging = config.logging();
@@ -51,9 +56,9 @@ public final Session newInstance( AccessMode mode, Bookmark bookmark )
     protected NetworkSession createSession( ConnectionProvider connectionProvider, RetryLogic retryLogic,
             AccessMode mode, Logging logging )
     {
-        return leakedSessionsLoggingEnabled ?
-               new LeakLoggingNetworkSession( connectionProvider, mode, retryLogic, logging ) :
-               new NetworkSession( connectionProvider, mode, retryLogic, logging );
+        return leakedSessionsLoggingEnabled
+               ? new LeakLoggingNetworkSession( connectionProvider, mode, retryLogic, eventExecutorGroup, logging )
+               : new NetworkSession( connectionProvider, mode, retryLogic, eventExecutorGroup, logging );
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/async/InternalPromise.java

@@ -20,6 +20,7 @@
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.concurrent.EventExecutorGroup;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.FutureListener;
 import io.netty.util.concurrent.GenericFutureListener;
@@ -40,7 +41,12 @@ public class InternalPromise<T> implements InternalFuture<T>, Promise<T>
 
     public InternalPromise( Bootstrap bootstrap )
     {
-        this( bootstrap.config().group().next() );
+        this( bootstrap.config().group() );
+    }
+
+    public InternalPromise( EventExecutorGroup eventExecutorGroup )
+    {
+        this( eventExecutorGroup.next() );
     }
 
     public InternalPromise( EventExecutor eventExecutor )

driver/src/main/java/org/neo4j/driver/internal/retry/ExponentialBackoffRetryLogic.java

@@ -124,7 +124,7 @@ public <T> T retry( Supplier<T> work )
     @Override
     public <T> InternalFuture<T> retryAsync( Supplier<InternalFuture<T>> work )
     {
-        InternalPromise<T> result = new InternalPromise<>( eventExecutorGroup.next() );
+        InternalPromise<T> result = new InternalPromise<>( eventExecutorGroup );
         executeWorkInEventLoop( result, work );
         return result;
     }