Merge pull request #401 from lutovich/1.7-eager-conn-init

Improve ordering of connection initialization

Author: lutovich

src/v1/driver.js

@@ -19,14 +19,15 @@
 
 import Session from './session';
 import Pool from './internal/pool';
-import {connect} from './internal/connector';
+import Connection from './internal/connection';
 import StreamObserver from './internal/stream-observer';
 import {newError, SERVICE_UNAVAILABLE} from './error';
 import {DirectConnectionProvider} from './internal/connection-providers';
 import Bookmark from './internal/bookmark';
 import ConnectivityVerifier from './internal/connectivity-verifier';
 import PoolConfig, {DEFAULT_ACQUISITION_TIMEOUT, DEFAULT_MAX_SIZE} from './internal/pool-config';
 import Logger from './internal/logger';
+import ConnectionErrorHandler from './internal/connection-error-handler';
 
 const DEFAULT_MAX_CONNECTION_LIFETIME = 60 * 60 * 1000; // 1 hour
 
@@ -62,18 +63,18 @@ class Driver {
    * @constructor
    * @param {string} hostPort
    * @param {string} userAgent
-   * @param {object} token
+   * @param {object} authToken
    * @param {object} config
    * @protected
    */
-  constructor(hostPort, userAgent, token = {}, config = {}) {
+  constructor(hostPort, userAgent, authToken = {}, config = {}) {
     sanitizeConfig(config);
 
     this._id = idGenerator++;
     this._hostPort = hostPort;
     this._userAgent = userAgent;
     this._openConnections = {};
-    this._token = token;
+    this._authToken = authToken;
     this._config = config;
     this._log = Logger.create(config);
     this._pool = new Pool(
@@ -127,18 +128,24 @@ class Driver {
   }
 
   /**
-   * Create a new connection instance.
-   * @return {Connection} new connector-api session instance, a low level session API.
+   * Create a new connection and initialize it.
+   * @return {Promise<Connection>} promise resolved with a new connection or rejected when failed to connect.
    * @access private
    */
   _createConnection(hostPort, release) {
-    const conn = connect(hostPort, this._config, this._connectionErrorCode(), this._log);
-    const streamObserver = new _ConnectionStreamObserver(this, conn);
-    conn.protocol().initialize(this._userAgent, this._token, streamObserver);
-    conn._release = () => release(hostPort, conn);
-
-    this._openConnections[conn.id] = conn;
-    return conn;
+    const connection = Connection.create(hostPort, this._config, this._createConnectionErrorHandler(), this._log);
+    connection._release = () => release(hostPort, connection);
+    this._openConnections[connection.id] = connection;
+
+    return connection.connect(this._userAgent, this._authToken)
+      .catch(error => {
+        if (this.onError) {
+          // notify Driver.onError callback about connection initialization errors
+          this.onError(error);
+        }
+        // propagate the error because connection failed to connect / initialize
+        throw error;
+      });
   }
 
   /**
@@ -186,7 +193,7 @@ class Driver {
     const sessionMode = Driver._validateSessionMode(mode);
     const connectionProvider = this._getOrCreateConnectionProvider();
     const bookmark = new Bookmark(bookmarkOrBookmarks);
-    return this._createSession(sessionMode, connectionProvider, bookmark, this._config);
+    return new Session(sessionMode, connectionProvider, bookmark, this._config);
   }
 
   static _validateSessionMode(rawMode) {
@@ -203,14 +210,8 @@ class Driver {
   }
 
   // Extension point
-  _createSession(mode, connectionProvider, bookmark, config) {
-    return new Session(mode, connectionProvider, bookmark, config);
-  }
-
-  // Extension point
-  _connectionErrorCode() {
-    // connection errors might result in different error codes depending on the driver
-    return SERVICE_UNAVAILABLE;
+  _createConnectionErrorHandler() {
+    return new ConnectionErrorHandler(SERVICE_UNAVAILABLE);
   }
 
   _getOrCreateConnectionProvider() {

src/v1/internal/connection-error-handler.js

@@ -0,0 +1,74 @@
+/**
+ * Copyright (c) 2002-2018 "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.
+ */
+
+import {SERVICE_UNAVAILABLE, SESSION_EXPIRED} from '../error';
+
+export default class ConnectionErrorHandler {
+
+  constructor(errorCode, handleUnavailability, handleWriteFailure) {
+    this._errorCode = errorCode;
+    this._handleUnavailability = handleUnavailability || noOpHandler;
+    this._handleWriteFailure = handleWriteFailure || noOpHandler;
+  }
+
+  /**
+   * Error code to use for network errors.
+   * @return {string} the error code.
+   */
+  errorCode() {
+    return this._errorCode;
+  }
+
+  /**
+   * Handle and transform the error.
+   * @param {Neo4jError} error the original error.
+   * @param {string} hostPort the host and port of the connection where the error happened.
+   * @return {Neo4jError} new error that should be propagated to the user.
+   */
+  handleAndTransformError(error, hostPort) {
+    if (isAvailabilityError(error)) {
+      return this._handleUnavailability(error, hostPort);
+    }
+    if (isFailureToWrite(error)) {
+      return this._handleWriteFailure(error, hostPort);
+    }
+    return error;
+  }
+}
+
+function isAvailabilityError(error) {
+  if (error) {
+    return error.code === SESSION_EXPIRED ||
+      error.code === SERVICE_UNAVAILABLE ||
+      error.code === 'Neo.TransientError.General.DatabaseUnavailable';
+  }
+  return false;
+}
+
+function isFailureToWrite(error) {
+  if (error) {
+    return error.code === 'Neo.ClientError.Cluster.NotALeader' ||
+      error.code === 'Neo.ClientError.General.ForbiddenOnReadOnlyDatabase';
+  }
+  return false;
+}
+
+function noOpHandler(error) {
+  return error;
+}

src/v1/internal/connection-holder.js

@@ -55,7 +55,7 @@ export default class ConnectionHolder {
   getConnection(streamObserver) {
     return this._connectionPromise.then(connection => {
       streamObserver.resolveConnection(connection);
-      return connection.initializationCompleted();
+      return connection;
     });
   }
 

src/v1/internal/connection-providers.js

@@ -26,6 +26,8 @@ import hasFeature from './features';
 import {DnsHostNameResolver, DummyHostNameResolver} from './host-name-resolvers';
 import RoutingUtil from './routing-util';
 
+const UNAUTHORIZED_ERROR_CODE = 'Neo.ClientError.Security.Unauthorized';
+
 class ConnectionProvider {
 
   acquireConnection(mode) {
@@ -195,20 +197,32 @@ export class LoadBalancer extends ConnectionProvider {
 
         // try next router
         return this._createSessionForRediscovery(currentRouter).then(session => {
-          return this._rediscovery.lookupRoutingTableOnRouter(session, currentRouter)
+          if (session) {
+            return this._rediscovery.lookupRoutingTableOnRouter(session, currentRouter);
+          } else {
+            // unable to acquire connection and create session towards the current router
+            // return null to signal that the next router should be tried
+            return null;
+          }
         });
       });
     }, Promise.resolve(null));
   }
 
   _createSessionForRediscovery(routerAddress) {
-    return this._connectionPool.acquire(routerAddress).then(connection => {
-      // initialized connection is required for routing procedure call
-      // server version needs to be known to decide which routing procedure to use
-      const initializedConnectionPromise = connection.initializationCompleted();
-      const connectionProvider = new SingleConnectionProvider(initializedConnectionPromise);
-      return new Session(READ, connectionProvider);
-    });
+    return this._connectionPool.acquire(routerAddress)
+      .then(connection => {
+        const connectionProvider = new SingleConnectionProvider(connection);
+        return new Session(READ, connectionProvider);
+      })
+      .catch(error => {
+        // unable to acquire connection towards the given router
+        if (error && error.code === UNAUTHORIZED_ERROR_CODE) {
+          // auth error is a sign of a configuration issue, rediscovery should not proceed
+          throw error;
+        }
+        return null;
+      });
   }
 
   _applyRoutingTableIfPossible(newRoutingTable) {
@@ -257,14 +271,14 @@ export class LoadBalancer extends ConnectionProvider {
 
 export class SingleConnectionProvider extends ConnectionProvider {
 
-  constructor(connectionPromise) {
+  constructor(connection) {
     super();
-    this._connectionPromise = connectionPromise;
+    this._connection = connection;
   }
 
   acquireConnection(mode) {
-    const connectionPromise = this._connectionPromise;
-    this._connectionPromise = null;
-    return connectionPromise;
+    const connection = this._connection;
+    this._connection = null;
+    return Promise.resolve(connection);
   }
 }