Expose transaction timeout and metadata in the API

This is the first draft of the API to allow transaction timeout and/or
metadata be set for explicit and auto-commit transactions. Both
settings can be specified in a `TransactionConfig` object which can be
created like this:

```
TransactionConfig config = TransactionConfig.builder()
            .withTimeout(Duration.ofSeconds(5))
            .withMetadata(Collections.singletonMap("key", "value"))
            .build();
```

Explicit transactions can take config like this:

```
Transaction tx = session.beginTransaction(config);
```

and auto-commit transactions accept it in various overloads of
`Session#run()` and `Session#runAsync()`:

```
session.run("CREATE ()", config);

session.runAsync("RETURN $x", Collections.singletonMap("x", 1) config);
```

Transaction functions accept config like this:

```
session.readTransaction(tx -> {...}, config);

session.writeTransactionAsync(tx -> {...}, config);
```

There also exists a possibility to specify default transaction
configuration per driver. It can be configured in the driver's config:

```
Config driverConfig = Config.build()
            .withDefaultTransactionConfig(txConfig)
            .toConfig();
```

More tests and javadocs will be added in subsequent commits.

Author: lutovich

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

@@ -22,6 +22,7 @@
 import java.util.concurrent.CompletionStage;
 
 import org.neo4j.driver.internal.types.InternalTypeSystem;
+import org.neo4j.driver.internal.util.Extract;
 import org.neo4j.driver.internal.value.MapValue;
 import org.neo4j.driver.v1.Record;
 import org.neo4j.driver.v1.Statement;
@@ -32,10 +33,6 @@
 import org.neo4j.driver.v1.Values;
 import org.neo4j.driver.v1.types.TypeSystem;
 
-import static org.neo4j.driver.internal.util.Extract.assertParameter;
-import static org.neo4j.driver.internal.util.Iterables.newHashMapWithSize;
-import static org.neo4j.driver.v1.Values.value;
-
 abstract class AbstractStatementRunner implements StatementRunner
 {
     @Override
@@ -103,14 +100,6 @@ private static Value parameters( Map<String,Object> map )
         {
             return Values.EmptyMap;
         }
-
-        Map<String,Value> asValues = newHashMapWithSize( map.size() );
-        for ( Map.Entry<String,Object> entry : map.entrySet() )
-        {
-            Object value = entry.getValue();
-            assertParameter( value );
-            asValues.put( entry.getKey(), value( value ) );
-        }
-        return new MapValue( asValues );
+        return new MapValue( Extract.mapOfValues( map ) );
     }
 }

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

@@ -31,6 +31,7 @@
 import org.neo4j.driver.v1.StatementResult;
 import org.neo4j.driver.v1.StatementResultCursor;
 import org.neo4j.driver.v1.Transaction;
+import org.neo4j.driver.v1.TransactionConfig;
 import org.neo4j.driver.v1.exceptions.ClientException;
 
 import static org.neo4j.driver.internal.util.Futures.completedWithNull;
@@ -78,9 +79,9 @@ public ExplicitTransaction( Connection connection, NetworkSession session )
         this.resultCursors = new ResultCursorsHolder();
     }
 
-    public CompletionStage<ExplicitTransaction> beginAsync( Bookmarks initialBookmarks )
+    public CompletionStage<ExplicitTransaction> beginAsync( Bookmarks initialBookmarks, TransactionConfig config )
     {
-        return protocol.beginTransaction( connection, initialBookmarks )
+        return protocol.beginTransaction( connection, initialBookmarks, config )
                 .handle( ( ignore, beginError ) ->
                 {
                     if ( beginError != null )

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

@@ -23,6 +23,7 @@
 import org.neo4j.driver.internal.util.Futures;
 import org.neo4j.driver.v1.AccessMode;
 import org.neo4j.driver.v1.Logging;
+import org.neo4j.driver.v1.TransactionConfig;
 
 import static java.lang.System.lineSeparator;
 
@@ -31,9 +32,9 @@ class LeakLoggingNetworkSession extends NetworkSession
     private final String stackTrace;
 
     LeakLoggingNetworkSession( ConnectionProvider connectionProvider, AccessMode mode, RetryLogic retryLogic,
-            Logging logging )
+            TransactionConfig defaultTransactionConfig, Logging logging )
     {
-        super( connectionProvider, mode, retryLogic, logging );
+        super( connectionProvider, mode, retryLogic, defaultTransactionConfig, logging );
         this.stackTrace = captureStackTrace();
     }
 

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

@@ -18,6 +18,7 @@
  */
 package org.neo4j.driver.internal;
 
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
@@ -36,9 +37,11 @@
 import org.neo4j.driver.v1.StatementResult;
 import org.neo4j.driver.v1.StatementResultCursor;
 import org.neo4j.driver.v1.Transaction;
+import org.neo4j.driver.v1.TransactionConfig;
 import org.neo4j.driver.v1.TransactionWork;
 import org.neo4j.driver.v1.exceptions.ClientException;
 
+import static java.util.Collections.emptyMap;
 import static java.util.concurrent.CompletableFuture.completedFuture;
 import static org.neo4j.driver.internal.util.Futures.completedWithNull;
 import static org.neo4j.driver.internal.util.Futures.failedFuture;
@@ -50,6 +53,7 @@ public class NetworkSession extends AbstractStatementRunner implements Session
     private final ConnectionProvider connectionProvider;
     private final AccessMode mode;
     private final RetryLogic retryLogic;
+    private final TransactionConfig defaultTransactionConfig;
     protected final Logger logger;
 
     private volatile Bookmarks bookmarks = Bookmarks.empty();
@@ -60,18 +64,37 @@ public class NetworkSession extends AbstractStatementRunner implements Session
     private final AtomicBoolean open = new AtomicBoolean( true );
 
     public NetworkSession( ConnectionProvider connectionProvider, AccessMode mode, RetryLogic retryLogic,
-            Logging logging )
+            TransactionConfig defaultTransactionConfig, Logging logging )
     {
         this.connectionProvider = connectionProvider;
         this.mode = mode;
         this.retryLogic = retryLogic;
+        this.defaultTransactionConfig = defaultTransactionConfig;
         this.logger = new PrefixedLogger( "[" + hashCode() + "]", logging.getLog( LOG_NAME ) );
     }
 
     @Override
     public StatementResult run( Statement statement )
     {
-        StatementResultCursor cursor = Futures.blockingGet( run( statement, false ),
+        return run( statement, TransactionConfig.empty() );
+    }
+
+    @Override
+    public StatementResult run( String statement, TransactionConfig config )
+    {
+        return run( statement, emptyMap(), config );
+    }
+
+    @Override
+    public StatementResult run( String statement, Map<String,Object> parameters, TransactionConfig config )
+    {
+        return run( new Statement( statement, parameters ), config );
+    }
+
+    @Override
+    public StatementResult run( Statement statement, TransactionConfig config )
+    {
+        StatementResultCursor cursor = Futures.blockingGet( run( statement, config, false ),
                 () -> terminateConnectionOnThreadInterrupt( "Thread interrupted while running query in session" ) );
 
         // query executed, it is safe to obtain a connection in a blocking way
@@ -81,9 +104,27 @@ public StatementResult run( Statement statement )
 
     @Override
     public CompletionStage<StatementResultCursor> runAsync( Statement statement )
+    {
+        return runAsync( statement, TransactionConfig.empty() );
+    }
+
+    @Override
+    public CompletionStage<StatementResultCursor> runAsync( String statement, TransactionConfig config )
+    {
+        return runAsync( statement, emptyMap(), config );
+    }
+
+    @Override
+    public CompletionStage<StatementResultCursor> runAsync( String statement, Map<String,Object> parameters, TransactionConfig config )
+    {
+        return runAsync( new Statement( statement, parameters ), config );
+    }
+
+    @Override
+    public CompletionStage<StatementResultCursor> runAsync( Statement statement, TransactionConfig config )
     {
         //noinspection unchecked
-        return (CompletionStage) run( statement, true );
+        return (CompletionStage) run( statement, config, true );
     }
 
     @Override
@@ -131,7 +172,13 @@ public CompletionStage<Void> closeAsync()
     @Override
     public Transaction beginTransaction()
     {
-        return beginTransaction( mode );
+        return beginTransaction( defaultTransactionConfig );
+    }
+
+    @Override
+    public Transaction beginTransaction( TransactionConfig config )
+    {
+        return beginTransaction( mode, config );
     }
 
     @Deprecated
@@ -144,33 +191,63 @@ public Transaction beginTransaction( String bookmark )
 
     @Override
     public CompletionStage<Transaction> beginTransactionAsync()
+    {
+        return beginTransactionAsync( defaultTransactionConfig );
+    }
+
+    @Override
+    public CompletionStage<Transaction> beginTransactionAsync( TransactionConfig config )
     {
         //noinspection unchecked
-        return (CompletionStage) beginTransactionAsync( mode );
+        return (CompletionStage) beginTransactionAsync( mode, config );
     }
 
     @Override
     public <T> T readTransaction( TransactionWork<T> work )
     {
-        return transaction( AccessMode.READ, work );
+        return readTransaction( work, defaultTransactionConfig );
+    }
+
+    @Override
+    public <T> T readTransaction( TransactionWork<T> work, TransactionConfig config )
+    {
+        return transaction( AccessMode.READ, work, config );
     }
 
     @Override
     public <T> CompletionStage<T> readTransactionAsync( TransactionWork<CompletionStage<T>> work )
     {
-        return transactionAsync( AccessMode.READ, work );
+        return readTransactionAsync( work, defaultTransactionConfig );
+    }
+
+    @Override
+    public <T> CompletionStage<T> readTransactionAsync( TransactionWork<CompletionStage<T>> work, TransactionConfig config )
+    {
+        return transactionAsync( AccessMode.READ, work, config );
     }
 
     @Override
     public <T> T writeTransaction( TransactionWork<T> work )
     {
-        return transaction( AccessMode.WRITE, work );
+        return writeTransaction( work, defaultTransactionConfig );
+    }
+
+    @Override
+    public <T> T writeTransaction( TransactionWork<T> work, TransactionConfig config )
+    {
+        return transaction( AccessMode.WRITE, work, config );
     }
 
     @Override
     public <T> CompletionStage<T> writeTransactionAsync( TransactionWork<CompletionStage<T>> work )
     {
-        return transactionAsync( AccessMode.WRITE, work );
+        return writeTransactionAsync( work, defaultTransactionConfig );
+    }
+
+    @Override
+    public <T> CompletionStage<T> writeTransactionAsync( TransactionWork<CompletionStage<T>> work, TransactionConfig config )
+    {
+        return transactionAsync( AccessMode.WRITE, work, config );
     }
 
     void setBookmarks( Bookmarks bookmarks )
@@ -225,15 +302,15 @@ CompletionStage<Boolean> currentConnectionIsOpen()
                 connection.isOpen() ); // and it's still open
     }
 
-    private <T> T transaction( AccessMode mode, TransactionWork<T> work )
+    private <T> T transaction( AccessMode mode, TransactionWork<T> work, TransactionConfig config )
     {
         // use different code path compared to async so that work is executed in the caller thread
         // caller thread will also be the one who sleeps between retries;
         // it is unsafe to execute retries in the event loop threads because this can cause a deadlock
         // event loop thread will bock and wait for itself to read some data
         return retryLogic.retry( () ->
         {
-            try ( Transaction tx = beginTransaction( mode ) )
+            try ( Transaction tx = beginTransaction( mode, config ) )
             {
                 try
                 {
@@ -252,12 +329,12 @@ private <T> T transaction( AccessMode mode, TransactionWork<T> work )
         } );
     }
 
-    private <T> CompletionStage<T> transactionAsync( AccessMode mode, TransactionWork<CompletionStage<T>> work )
+    private <T> CompletionStage<T> transactionAsync( AccessMode mode, TransactionWork<CompletionStage<T>> work, TransactionConfig config )
     {
         return retryLogic.retryAsync( () ->
         {
             CompletableFuture<T> resultFuture = new CompletableFuture<>();
-            CompletionStage<ExplicitTransaction> txFuture = beginTransactionAsync( mode );
+            CompletionStage<ExplicitTransaction> txFuture = beginTransactionAsync( mode, config );
 
             txFuture.whenComplete( ( tx, completionError ) ->
             {
@@ -358,26 +435,27 @@ private <T> void closeTxAfterSucceededTransactionWork( ExplicitTransaction tx, C
         }
     }
 
-    private CompletionStage<InternalStatementResultCursor> run( Statement statement, boolean waitForRunResponse )
+    private CompletionStage<InternalStatementResultCursor> run( Statement statement, TransactionConfig config, boolean waitForRunResponse )
     {
         ensureSessionIsOpen();
 
         CompletionStage<InternalStatementResultCursor> newResultCursorStage = ensureNoOpenTxBeforeRunningQuery()
                 .thenCompose( ignore -> acquireConnection( mode ) )
-                .thenCompose( connection -> connection.protocol().runInAutoCommitTransaction( connection, statement, waitForRunResponse ) );
+                .thenCompose( connection ->
+                        connection.protocol().runInAutoCommitTransaction( connection, statement, bookmarks, config, waitForRunResponse ) );
 
         resultCursorStage = newResultCursorStage.exceptionally( error -> null );
 
         return newResultCursorStage;
     }
 
-    private Transaction beginTransaction( AccessMode mode )
+    private Transaction beginTransaction( AccessMode mode, TransactionConfig config )
     {
-        return Futures.blockingGet( beginTransactionAsync( mode ),
+        return Futures.blockingGet( beginTransactionAsync( mode, config ),
                 () -> terminateConnectionOnThreadInterrupt( "Thread interrupted while starting a transaction" ) );
     }
 
-    private CompletionStage<ExplicitTransaction> beginTransactionAsync( AccessMode mode )
+    private CompletionStage<ExplicitTransaction> beginTransactionAsync( AccessMode mode, TransactionConfig config )
     {
         ensureSessionIsOpen();
 
@@ -387,7 +465,7 @@ private CompletionStage<ExplicitTransaction> beginTransactionAsync( AccessMode m
                 .thenCompose( connection ->
                 {
                     ExplicitTransaction tx = new ExplicitTransaction( connection, NetworkSession.this );
-                    return tx.beginAsync( bookmarks );
+                    return tx.beginAsync( bookmarks, config );
                 } );
 
         // update the reference to the only known transaction

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

@@ -26,11 +26,13 @@
 import org.neo4j.driver.v1.Config;
 import org.neo4j.driver.v1.Logging;
 import org.neo4j.driver.v1.Session;
+import org.neo4j.driver.v1.TransactionConfig;
 
 public class SessionFactoryImpl implements SessionFactory
 {
     private final ConnectionProvider connectionProvider;
     private final RetryLogic retryLogic;
+    private final TransactionConfig defaultTransactionConfig;
     private final Logging logging;
     private final boolean leakedSessionsLoggingEnabled;
 
@@ -39,6 +41,7 @@ public class SessionFactoryImpl implements SessionFactory
         this.connectionProvider = connectionProvider;
         this.leakedSessionsLoggingEnabled = config.logLeakedSessions();
         this.retryLogic = retryLogic;
+        this.defaultTransactionConfig = config.defaultTransactionConfig();
         this.logging = config.logging();
     }
 
@@ -78,7 +81,7 @@ private NetworkSession createSession( ConnectionProvider connectionProvider, Ret
             AccessMode mode, Logging logging )
     {
         return leakedSessionsLoggingEnabled
-               ? new LeakLoggingNetworkSession( connectionProvider, mode, retryLogic, logging )
-               : new NetworkSession( connectionProvider, mode, retryLogic, logging );
+               ? new LeakLoggingNetworkSession( connectionProvider, mode, retryLogic, defaultTransactionConfig, logging )
+               : new NetworkSession( connectionProvider, mode, retryLogic, defaultTransactionConfig, logging );
     }
 }

driver/src/main/java/org/neo4j/driver/internal/cluster/RoutingProcedureRunner.java

@@ -22,12 +22,14 @@
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
 
+import org.neo4j.driver.internal.Bookmarks;
 import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.internal.util.Futures;
 import org.neo4j.driver.internal.util.ServerVersion;
 import org.neo4j.driver.v1.Record;
 import org.neo4j.driver.v1.Statement;
 import org.neo4j.driver.v1.StatementResultCursor;
+import org.neo4j.driver.v1.TransactionConfig;
 import org.neo4j.driver.v1.exceptions.ClientException;
 
 import static org.neo4j.driver.internal.util.ServerVersion.v3_2_0;
@@ -60,7 +62,7 @@ public CompletionStage<RoutingProcedureResponse> run( CompletionStage<Connection
     CompletionStage<List<Record>> runProcedure( Connection connection, Statement procedure )
     {
         return connection.protocol()
-                .runInAutoCommitTransaction( connection, procedure, true )
+                .runInAutoCommitTransaction( connection, procedure, Bookmarks.empty(), TransactionConfig.empty(), true )
                 .thenCompose( StatementResultCursor::listAsync );
     }