Draft of async API

driver/pom.xml

@@ -8,6 +8,7 @@
     <build.revision></build.revision>
     <bundle.name>${project.groupId}.${project.artifactId}</bundle.name>
     <maven.build.timestamp.format>'v'yyyyMMdd-HHmm</maven.build.timestamp.format>
+      <netty.version>4.1.15.Final</netty.version>
   </properties>
 
   <parent>
@@ -32,7 +33,13 @@
   </licenses>
 
   <dependencies>
-    <!-- Test dependencies -->
+      <dependency>
+          <groupId>io.netty</groupId>
+          <artifactId>netty-all</artifactId>
+          <version>${netty.version}</version>
+      </dependency>
+
+      <!-- Test dependencies -->
     <dependency>
       <groupId>org.hamcrest</groupId>
       <artifactId>hamcrest-library</artifactId>

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

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2002-2017 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.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;
+
+public interface ResultResourcesHandler
+{
+    void resultFetched();
+
+    void resultFailed( Throwable error );
+
+    ResultResourcesHandler NO_OP = new ResultResourcesHandler()
+    {
+        @Override
+        public void resultFetched()
+        {
+        }
+
+        @Override
+        public void resultFailed( Throwable error )
+        {
+        }
+    };
+}

driver/src/main/java/org/neo4j/driver/internal/BookmarkCollector.java → driver/src/main/java/org/neo4j/driver/StatementKeys.java

@@ -16,23 +16,27 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.neo4j.driver.internal;
+package org.neo4j.driver;
 
-import org.neo4j.driver.internal.spi.Collector.NoOperationCollector;
+import java.util.Collections;
+import java.util.List;
 
-class BookmarkCollector extends NoOperationCollector
+public class StatementKeys
 {
-    private final ExplicitTransaction transaction;
+    private List<String> keys;
 
-    BookmarkCollector( ExplicitTransaction transaction )
+    public boolean isPopulated()
     {
-        this.transaction = transaction;
+        return keys != null;
     }
 
-    @Override
-    public void bookmark( Bookmark bookmark )
+    public void set( List<String> keys )
     {
-        transaction.setBookmark( bookmark );
+        this.keys = keys;
     }
 
+    public List<String> asList()
+    {
+        return keys == null ? Collections.<String>emptyList() : keys;
+    }
 }

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

@@ -52,18 +52,18 @@ private static String driverVersion()
 
     private final AuthToken authToken;
     private final String userAgent;
-    private final int timeoutMillis;
+    private final int connectTimeoutMillis;
 
-    public ConnectionSettings( AuthToken authToken, String userAgent, int timeoutMillis )
+    public ConnectionSettings( AuthToken authToken, String userAgent, int connectTimeoutMillis )
     {
         this.authToken = authToken;
         this.userAgent = userAgent;
-        this.timeoutMillis = timeoutMillis;
+        this.connectTimeoutMillis = connectTimeoutMillis;
     }
 
-    public ConnectionSettings( AuthToken authToken, int timeoutMillis )
+    public ConnectionSettings( AuthToken authToken, int connectTimeoutMillis )
     {
-        this( authToken, DEFAULT_USER_AGENT, timeoutMillis );
+        this( authToken, DEFAULT_USER_AGENT, connectTimeoutMillis );
     }
 
     public AuthToken authToken()
@@ -76,8 +76,8 @@ public String userAgent()
         return userAgent;
     }
 
-    public int timeoutMillis()
+    public int connectTimeoutMillis()
     {
-        return timeoutMillis;
+        return connectTimeoutMillis;
     }
 }

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

@@ -18,6 +18,9 @@
  */
 package org.neo4j.driver.internal;
 
+import org.neo4j.driver.internal.async.AsyncConnection;
+import org.neo4j.driver.internal.async.InternalFuture;
+import org.neo4j.driver.internal.async.pool.AsyncConnectionPool;
 import org.neo4j.driver.internal.net.BoltServerAddress;
 import org.neo4j.driver.internal.spi.ConnectionPool;
 import org.neo4j.driver.internal.spi.ConnectionProvider;
@@ -32,11 +35,13 @@ public class DirectConnectionProvider implements ConnectionProvider
 {
     private final BoltServerAddress address;
     private final ConnectionPool pool;
+    private final AsyncConnectionPool asyncPool;
 
-    DirectConnectionProvider( BoltServerAddress address, ConnectionPool pool )
+    DirectConnectionProvider( BoltServerAddress address, ConnectionPool pool, AsyncConnectionPool asyncPool )
     {
         this.address = address;
         this.pool = pool;
+        this.asyncPool = asyncPool;
 
         verifyConnectivity();
     }
@@ -47,10 +52,17 @@ public PooledConnection acquireConnection( AccessMode mode )
         return pool.acquire( address );
     }
 
+    @Override
+    public InternalFuture<AsyncConnection> acquireAsyncConnection( AccessMode mode )
+    {
+        return asyncPool.acquire( address );
+    }
+
     @Override
     public void close() throws Exception
     {
         pool.close();
+        asyncPool.closeAsync().syncUninterruptibly();
     }
 
     public BoltServerAddress getAddress()

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

@@ -18,10 +18,17 @@
  */
 package org.neo4j.driver.internal;
 
+import io.netty.bootstrap.Bootstrap;
+
 import java.io.IOException;
 import java.net.URI;
 import java.security.GeneralSecurityException;
 
+import org.neo4j.driver.internal.async.AsyncConnectorImpl;
+import org.neo4j.driver.internal.async.BootstrapFactory;
+import org.neo4j.driver.internal.async.pool.ActiveChannelTracker;
+import org.neo4j.driver.internal.async.pool.AsyncConnectionPool;
+import org.neo4j.driver.internal.async.pool.AsyncConnectionPoolImpl;
 import org.neo4j.driver.internal.cluster.RoutingContext;
 import org.neo4j.driver.internal.cluster.RoutingSettings;
 import org.neo4j.driver.internal.cluster.loadbalancing.LeastConnectedLoadBalancingStrategy;
@@ -59,22 +66,28 @@ public class DriverFactory
     public final Driver newInstance( URI uri, AuthToken authToken, RoutingSettings routingSettings,
             RetrySettings retrySettings, Config config )
     {
+        authToken = authToken == null ? AuthTokens.none() : authToken;
+
         BoltServerAddress address = new BoltServerAddress( uri );
         RoutingSettings newRoutingSettings = routingSettings.withRoutingContext( new RoutingContext( uri ) );
         SecurityPlan securityPlan = createSecurityPlan( address, config );
         ConnectionPool connectionPool = createConnectionPool( authToken, securityPlan, config );
         RetryLogic retryLogic = createRetryLogic( retrySettings, config.logging() );
 
+        AsyncConnectionPool asyncConnectionPool = createAsyncConnectionPool( authToken, securityPlan, config );
+
         try
         {
-            return createDriver( uri, address, connectionPool, config, newRoutingSettings, securityPlan, retryLogic );
+            return createDriver( uri, address, connectionPool, config, newRoutingSettings, securityPlan, retryLogic,
+                    asyncConnectionPool );
         }
         catch ( Throwable driverError )
         {
             // we need to close the connection pool if driver creation threw exception
             try
             {
                 connectionPool.close();
+                asyncConnectionPool.closeAsync().syncUninterruptibly();
             }
             catch ( Throwable closeError )
             {
@@ -84,16 +97,33 @@ public final Driver newInstance( URI uri, AuthToken authToken, RoutingSettings r
         }
     }
 
+    private AsyncConnectionPool createAsyncConnectionPool( AuthToken authToken, SecurityPlan securityPlan,
+            Config config )
+    {
+        Clock clock = createClock();
+        ConnectionSettings connectionSettings = new ConnectionSettings( authToken, config.connectionTimeoutMillis() );
+        ActiveChannelTracker activeChannelTracker = new ActiveChannelTracker( config.logging() );
+        AsyncConnectorImpl connector = new AsyncConnectorImpl( connectionSettings, securityPlan,
+                activeChannelTracker, config.logging(), clock );
+        Bootstrap bootstrap = BootstrapFactory.newBootstrap();
+        PoolSettings poolSettings = new PoolSettings( config.maxIdleConnectionPoolSize(),
+                config.idleTimeBeforeConnectionTest(), config.maxConnectionLifetimeMillis(),
+                config.maxConnectionPoolSize(),
+                config.connectionAcquisitionTimeoutMillis() );
+        return new AsyncConnectionPoolImpl( connector, bootstrap, activeChannelTracker, poolSettings, config.logging(),
+                clock );
+    }
+
     private Driver createDriver( URI uri, BoltServerAddress address, ConnectionPool connectionPool,
             Config config, RoutingSettings routingSettings, SecurityPlan securityPlan,
-            RetryLogic retryLogic )
+            RetryLogic retryLogic, AsyncConnectionPool asyncConnectionPool )
     {
         String scheme = uri.getScheme().toLowerCase();
         switch ( scheme )
         {
         case BOLT_URI_SCHEME:
             assertNoRoutingContext( uri, routingSettings );
-            return createDirectDriver( address, connectionPool, config, securityPlan, retryLogic );
+            return createDirectDriver( address, connectionPool, config, securityPlan, retryLogic, asyncConnectionPool );
         case BOLT_ROUTING_URI_SCHEME:
             return createRoutingDriver( address, connectionPool, config, routingSettings, securityPlan, retryLogic );
         default:
@@ -107,9 +137,10 @@ 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 )
+            SecurityPlan securityPlan, RetryLogic retryLogic, AsyncConnectionPool asyncConnectionPool )
     {
-        ConnectionProvider connectionProvider = new DirectConnectionProvider( address, connectionPool );
+        ConnectionProvider connectionProvider =
+                new DirectConnectionProvider( address, connectionPool, asyncConnectionPool );
         SessionFactory sessionFactory = createSessionFactory( connectionProvider, retryLogic, config );
         return createDriver( config, securityPlan, sessionFactory );
     }
@@ -173,11 +204,10 @@ private static LoadBalancingStrategy createLoadBalancingStrategy( Config config,
      */
     protected ConnectionPool createConnectionPool( AuthToken authToken, SecurityPlan securityPlan, Config config )
     {
-        authToken = authToken == null ? AuthTokens.none() : authToken;
-
         ConnectionSettings connectionSettings = new ConnectionSettings( authToken, config.connectionTimeoutMillis() );
         PoolSettings poolSettings = new PoolSettings( config.maxIdleConnectionPoolSize(),
-                config.idleTimeBeforeConnectionTest(), config.maxConnectionLifetime() );
+                config.idleTimeBeforeConnectionTest(), config.maxConnectionLifetimeMillis(),
+                config.maxConnectionPoolSize(), config.connectionAcquisitionTimeoutMillis() );
         Connector connector = createConnector( connectionSettings, securityPlan, config.logging() );
 
         return new SocketConnectionPool( poolSettings, connector, createClock(), config.logging() );
@@ -198,7 +228,7 @@ protected Clock createClock()
      * <p>
      * <b>This method is protected only for testing</b>
      */
-    protected Connector createConnector( ConnectionSettings connectionSettings, SecurityPlan securityPlan,
+    protected Connector createConnector( final ConnectionSettings connectionSettings, SecurityPlan securityPlan,
             Logging logging )
     {
         return new SocketConnector( connectionSettings, securityPlan, logging );