Internal refactoring, extra tests and Driver.servers

Author: technige

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

@@ -18,21 +18,22 @@
  */
 package org.neo4j.driver.internal;
 
-import org.neo4j.driver.internal.connector.socket.SocketConnector;
-import org.neo4j.driver.internal.pool.InternalConnectionPool;
+import org.neo4j.driver.internal.pool.SocketConnectionPool;
 import org.neo4j.driver.internal.pool.PoolSettings;
 import org.neo4j.driver.internal.security.SecurityPlan;
 import org.neo4j.driver.internal.spi.ConnectionPool;
 import org.neo4j.driver.internal.util.BoltServerAddress;
-import org.neo4j.driver.internal.util.Clock;
 import org.neo4j.driver.v1.Driver;
 import org.neo4j.driver.v1.Logging;
 import org.neo4j.driver.v1.Session;
 import org.neo4j.driver.v1.exceptions.ClientException;
 import org.neo4j.driver.v1.exceptions.Neo4jException;
 
+import java.util.*;
+
 public class DirectDriver implements Driver
 {
+    private final Set<BoltServerAddress> servers;
     private final BoltServerAddress address;
     private final SecurityPlan securityPlan;
     private final Logging logging;
@@ -41,9 +42,18 @@ public class DirectDriver implements Driver
     public DirectDriver( BoltServerAddress address, SecurityPlan securityPlan, PoolSettings poolSettings, Logging logging )
     {
         this.address = address;
+        Set<BoltServerAddress> servers = new HashSet<>();
+        servers.add( this.address );
+        this.servers = Collections.unmodifiableSet( servers );
         this.securityPlan = securityPlan;
         this.logging = logging;
-        this.connections = new InternalConnectionPool( new SocketConnector(), Clock.SYSTEM, securityPlan, poolSettings, logging );
+        this.connections = new SocketConnectionPool( securityPlan, poolSettings, logging );
+    }
+
+    @Override
+    public Set<BoltServerAddress> servers()
+    {
+        return servers;
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/connector/socket/SocketClient.java

@@ -170,7 +170,7 @@ public boolean isOpen()
 
     private SocketProtocol negotiateProtocol() throws IOException
     {
-        logger.debug( "~~ [HANDSHAKE] [0x6060B017, 1, 0, 0, 0]." );
+        logger.debug( "~~ [HANDSHAKE] [0x6060B017, 1, 0, 0, 0]" );
         //Propose protocol versions
         ByteBuffer buf = ByteBuffer.allocateDirect( 5 * 4 ).order( BIG_ENDIAN );
         buf.putInt( MAGIC_PREAMBLE );
@@ -240,7 +240,7 @@ public static ByteChannel create( BoltServerAddress address, SecurityPlan securi
             SocketChannel soChannel = SocketChannel.open();
             soChannel.setOption( StandardSocketOptions.SO_REUSEADDR, true );
             soChannel.setOption( StandardSocketOptions.SO_KEEPALIVE, true );
-            soChannel.connect( new InetSocketAddress( address.host(), address.port() ) );
+            soChannel.connect( address.toSocketAddress() );
 
             ByteChannel channel;
 

driver/src/main/java/org/neo4j/driver/internal/connector/socket/SocketConnector.java

@@ -18,20 +18,26 @@
  */
 package org.neo4j.driver.internal.pool;
 
-import java.net.URI;
-import java.util.Arrays;
+import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.neo4j.driver.internal.Version;
+import org.neo4j.driver.internal.connector.ConcurrencyGuardingConnection;
+import org.neo4j.driver.internal.connector.socket.SocketConnection;
+import org.neo4j.driver.internal.security.InternalAuthToken;
 import org.neo4j.driver.internal.security.SecurityPlan;
 import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.internal.spi.ConnectionPool;
-import org.neo4j.driver.internal.spi.Connector;
 import org.neo4j.driver.internal.util.BoltServerAddress;
 import org.neo4j.driver.internal.util.Clock;
+import org.neo4j.driver.v1.AuthToken;
+import org.neo4j.driver.v1.AuthTokens;
 import org.neo4j.driver.v1.Logging;
+import org.neo4j.driver.v1.Value;
+import org.neo4j.driver.v1.exceptions.ClientException;
 import org.neo4j.driver.v1.exceptions.Neo4jException;
 
 /**
@@ -46,31 +52,52 @@
  * The driver is thread safe. Each thread could try to get a session from the pool and then return it to the pool
  * at the same time.
  */
-public class InternalConnectionPool implements ConnectionPool
+public class SocketConnectionPool implements ConnectionPool
 {
 
     /**
-     * Pools, organized by URL.
+     * Pools, organized by server address.
      */
     private final ConcurrentHashMap<BoltServerAddress,BlockingQueue<PooledConnection>> pools = new ConcurrentHashMap<>();
 
-    private final Connector connector;
+    private final Clock clock = Clock.SYSTEM;
+
     private final SecurityPlan securityPlan;
-    private final Clock clock;
     private final PoolSettings poolSettings;
     private final Logging logging;
 
     /** Shutdown flag */
     private final AtomicBoolean stopped = new AtomicBoolean( false );
 
-    public InternalConnectionPool( Connector connector, Clock clock, SecurityPlan securityPlan,
-                                   PoolSettings poolSettings, Logging logging )
+    public SocketConnectionPool( SecurityPlan securityPlan, PoolSettings poolSettings, Logging logging )
     {
         this.securityPlan = securityPlan;
-        this.clock = clock;
         this.poolSettings = poolSettings;
         this.logging = logging;
-        this.connector = connector;
+    }
+
+    private Connection connect( BoltServerAddress address ) throws ClientException
+    {
+        Connection conn = new SocketConnection( address, securityPlan, logging );
+
+        // Because SocketConnection is not thread safe, wrap it in this guard
+        // to ensure concurrent access leads causes application errors
+        conn = new ConcurrencyGuardingConnection( conn );
+        conn.init( "bolt-java-driver/" + Version.driverVersion(), tokenAsMap( securityPlan.authToken() ) );
+        return conn;
+    }
+
+    private static Map<String,Value> tokenAsMap( AuthToken token )
+    {
+        if( token instanceof InternalAuthToken )
+        {
+            return ((InternalAuthToken) token).toMap();
+        }
+        else
+        {
+            throw new ClientException( "Unknown authentication token, `" + token + "`. Please use one of the supported " +
+                    "tokens from `" + AuthTokens.class.getSimpleName() + "`." );
+        }
     }
 
     @Override
@@ -84,8 +111,8 @@ public Connection acquire( BoltServerAddress address )
         PooledConnection conn = connections.poll();
         if ( conn == null )
         {
-            conn = new PooledConnection(connector.connect( address, securityPlan, logging ), new
-                    PooledConnectionReleaseConsumer( connections, stopped, poolSettings ), clock);
+            conn = new PooledConnection( connect( address ), new
+                    PooledConnectionReleaseConsumer( connections, stopped, poolSettings ), clock );
         }
         conn.updateUsageTimestamp();
         return conn;

…nternal/pool/InternalConnectionPool.java …/internal/pool/SocketConnectionPool.javadriver/src/main/java/org/neo4j/driver/internal/pool/InternalConnectionPool.java renamed to driver/src/main/java/org/neo4j/driver/internal/pool/SocketConnectionPool.java driver/src/main/java/org/neo4j/driver/internal/pool/InternalConnectionPool.java renamed to driver/src/main/java/org/neo4j/driver/internal/pool/SocketConnectionPool.java

@@ -16,17 +16,19 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.neo4j.driver.internal.spi;
 
+package org.neo4j.driver.internal.spi;
 
 import org.neo4j.driver.internal.util.BoltServerAddress;
 
 public interface ConnectionPool extends AutoCloseable
 {
     /**
-     * Acquire a connection - if a live connection exists in the pool, it will be used, otherwise a new connection
-     * is created with an applicable {@link Connector}.
+     * Acquire a connection - if a live connection exists in the pool, it will
+     * be used, otherwise a new connection will be created.
+     *
      * @param address
      */
     Connection acquire( BoltServerAddress address );
+
 }

driver/src/main/java/org/neo4j/driver/internal/spi/ConnectionPool.java

@@ -44,6 +44,8 @@ public static BoltServerAddress from( URI uri )
     private final String host;
     private final int port;
 
+    private SocketAddress socketAddress = null;  // created lazily if required
+
     public BoltServerAddress( String host, int port )
     {
         this.host = host;
@@ -55,12 +57,61 @@ public BoltServerAddress( String host )
         this( host, DEFAULT_PORT );
     }
 
+    @Override
+    public boolean equals( Object obj )
+    {
+        if ( this == obj )
+        {
+            return true;
+        }
+        if ( !(obj instanceof BoltServerAddress) )
+        {
+            return false;
+        }
+        BoltServerAddress address = (BoltServerAddress) obj;
+        return host.equals( address.host ) && port == address.port;
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return 31 * host.hashCode() + port;
+    }
+
     @Override
     public String toString()
     {
         return format( "%s:%d", host, port );
     }
 
+    public SocketAddress toSocketAddress()
+    {
+        if (socketAddress == null)
+        {
+            socketAddress = new InetSocketAddress( host, port );
+        }
+        return socketAddress;
+    }
+
+    /**
+     * Resolve the host name down to an IP address, if not already resolved.
+     *
+     * @return this instance if already resolved, otherwise a new address instance
+     * @throws UnknownHostException
+     */
+    public BoltServerAddress resolve() throws UnknownHostException
+    {
+        String hostAddress = InetAddress.getByName( host ).getHostAddress();
+        if ( hostAddress.equals( host ) )
+        {
+            return this;
+        }
+        else
+        {
+            return new BoltServerAddress( hostAddress, port );
+        }
+    }
+
     public String host()
     {
         return host;