Lazy connection initialization and eager release

This commit effectively decouples session from connection. Session now
holds two `ConnectionHolder`s that lazily pick connection from the pool
when it is needed and try to release connection back to the pool when it
does not have any users. Session keeps one connection holder for read
connection and one for write connection.

Connection is acquired for `Session#run()` and released when the
returned result is either fully consumed or error happens. Also
connection is acquired for `Session#beginTransaction()` and released
when transaction commits or rolls back.

Author: lutovich

src/v1/internal/connection-holder.js

@@ -0,0 +1,137 @@
+/**
+ * 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.
+ */
+
+import {newError} from '../error';
+
+/**
+ * Utility to lazily initialize connections and return them back to the pool when unused.
+ */
+export default class ConnectionHolder {
+
+  /**
+   * @constructor
+   * @param {string} mode - the access mode for new connection holder.
+   * @param {ConnectionProvider} connectionProvider - the connection provider to acquire connections from.
+   */
+  constructor(mode, connectionProvider) {
+    this._mode = mode;
+    this._connectionProvider = connectionProvider;
+    this._referenceCount = 0;
+    this._connectionPromise = Promise.resolve(null);
+  }
+
+  /**
+   * Make this holder initialize new connection if none exists already.
+   * @return {undefined}
+   */
+  initializeConnection() {
+    if (this._referenceCount === 0) {
+      this._connectionPromise = this._connectionProvider.acquireConnection(this._mode);
+    }
+    this._referenceCount++;
+  }
+
+  /**
+   * Get the current connection promise.
+   * @return {Promise<Connection>} promise resolved with the current connection.
+   */
+  getConnection() {
+    return this._connectionPromise;
+  }
+
+  /**
+   * Notify this holder that single party does not require current connection any more.
+   * @return {Promise<Connection>} promise resolved with the current connection.
+   */
+  releaseConnection() {
+    if (this._referenceCount === 0) {
+      return this._connectionPromise;
+    }
+
+    this._referenceCount--;
+    if (this._referenceCount === 0) {
+      // release a connection without muting ACK_FAILURE, this is the last action on this connection
+      return this._releaseConnection(true);
+    }
+    return this._connectionPromise;
+  }
+
+  /**
+   * Closes this holder and releases current connection (if any) despite any existing users.
+   * @return {Promise<Connection>} promise resolved when current connection is released to the pool.
+   */
+  close() {
+    if (this._referenceCount === 0) {
+      return this._connectionPromise;
+    }
+    this._referenceCount = 0;
+    // release a connection and mute ACK_FAILURE, this might be called concurrently with other
+    // operations and thus should ignore failure handling
+    return this._releaseConnection(false);
+  }
+
+  /**
+   * Return the current pooled connection instance to the connection pool.
+   * We don't pool Session instances, to avoid users using the Session after they've called close.
+   * The `Session` object is just a thin wrapper around Connection anyway, so it makes little difference.
+   * @return {Promise} - promise resolved then connection is returned to the pool.
+   * @private
+   */
+  _releaseConnection(sync) {
+    this._connectionPromise = this._connectionPromise.then(connection => {
+      if (connection) {
+        if(sync) {
+          connection.reset();
+        } else {
+          connection.resetAsync();
+        }
+        connection.sync();
+        connection._release();
+      }
+    }).catch(ignoredError => {
+    });
+
+    return this._connectionPromise;
+  }
+}
+
+class EmptyConnectionHolder extends ConnectionHolder {
+
+  initializeConnection() {
+    // nothing to initialize
+  }
+
+  getConnection() {
+    return Promise.reject(newError('This connection holder does not serve connections'));
+  }
+
+  releaseConnection() {
+    return Promise.resolve();
+  }
+
+  close() {
+    return Promise.resolve();
+  }
+}
+
+/**
+ * Connection holder that does not manage any connections.
+ * @type {ConnectionHolder}
+ */
+export const EMPTY_CONNECTION_HOLDER = new EmptyConnectionHolder();

src/v1/internal/connection-providers.js

@@ -33,7 +33,9 @@ class ConnectionProvider {
   _withAdditionalOnErrorCallback(connectionPromise, driverOnErrorCallback) {
     // install error handler from the driver on the connection promise; this callback is installed separately
     // so that it does not handle errors, instead it is just an additional error reporting facility.
-    connectionPromise.catch(error => driverOnErrorCallback(error));
+    connectionPromise.catch(error => {
+      driverOnErrorCallback(error)
+    });
     // return the original connection promise
     return connectionPromise;
   }
@@ -49,7 +51,8 @@ export class DirectConnectionProvider extends ConnectionProvider {
   }
 
   acquireConnection(mode) {
-    const connectionPromise = Promise.resolve(this._connectionPool.acquire(this._address));
+    const connection = this._connectionPool.acquire(this._address);
+    const connectionPromise = Promise.resolve(connection);
     return this._withAdditionalOnErrorCallback(connectionPromise, this._driverOnErrorCallback);
   }
 }

src/v1/internal/connector.js

@@ -96,7 +96,6 @@ function log(actor, msg) {
   }
 }
 
-
 function NO_OP(){}
 
 let NO_OP_OBSERVER = {
@@ -384,9 +383,9 @@ class Connection {
     this._chunker.messageBoundary();
   }
 
-  /** Queue a RESET-message to be sent to the database */
-  reset( observer ) {
-    log("C", "RESET");
+  /** Queue a RESET-message to be sent to the database. Mutes failure handling. */
+  resetAsync( observer ) {
+    log("C", "RESET_ASYNC");
     this._isHandlingFailure = true;
     let self = this;
     let wrappedObs = {
@@ -404,6 +403,14 @@ class Connection {
     this._chunker.messageBoundary();
   }
 
+  /** Queue a RESET-message to be sent to the database */
+  reset(observer) {
+    log('C', 'RESET');
+    this._queueObserver(observer);
+    this._packer.packStruct(RESET, [], (err) => this._handleFatalError(err));
+    this._chunker.messageBoundary();
+  }
+
   /** Queue a ACK_FAILURE-message to be sent to the database */
   _ackFailure( observer ) {
     log("C", "ACK_FAILURE");

src/v1/internal/get-servers-util.js

@@ -17,9 +17,9 @@
  * limitations under the License.
  */
 
-import RoundRobinArray from "./round-robin-array";
-import {newError, PROTOCOL_ERROR, SERVICE_UNAVAILABLE} from "../error";
-import Integer, {int} from "../integer";
+import RoundRobinArray from './round-robin-array';
+import {newError, PROTOCOL_ERROR, SERVICE_UNAVAILABLE} from '../error';
+import Integer, {int} from '../integer';
 
 const PROCEDURE_CALL = 'CALL dbms.cluster.routing.getServers';
 const PROCEDURE_NOT_FOUND_CODE = 'Neo.ClientError.Procedure.ProcedureNotFound';

src/v1/result.js

@@ -18,6 +18,7 @@
  */
 
 import ResultSummary from './result-summary';
+import {EMPTY_CONNECTION_HOLDER} from './internal/connection-holder';
 
 /**
   * A stream of {@link Record} representing the result of a statement.
@@ -32,13 +33,15 @@ class Result {
    * @param {mixed} statement - Cypher statement to execute
    * @param {Object} parameters - Map with parameters to use in statement
    * @param metaSupplier function, when called provides metadata
+   * @param {ConnectionHolder} connectionHolder - to be notified when result is either fully consumed or error happened.
    */
-  constructor(streamObserver, statement, parameters, metaSupplier) {
+  constructor(streamObserver, statement, parameters, metaSupplier, connectionHolder) {
     this._streamObserver = streamObserver;
     this._p = null;
     this._statement = statement;
     this._parameters = parameters || {};
     this._metaSupplier = metaSupplier || function(){return {};};
+    this._connectionHolder = connectionHolder || EMPTY_CONNECTION_HOLDER;
   }
 
   /**
@@ -99,23 +102,39 @@ class Result {
    * @return
    */
   subscribe(observer) {
-    let onCompletedOriginal = observer.onCompleted;
-    let self = this;
-    let onCompletedWrapper = (metadata) => {
+    const onCompletedOriginal = observer.onCompleted;
+    const self = this;
+    const onCompletedWrapper = (metadata) => {
 
-      let additionalMeta = self._metaSupplier();
-      for(var key in additionalMeta) {
+      const additionalMeta = self._metaSupplier();
+      for(let key in additionalMeta) {
         if (additionalMeta.hasOwnProperty(key)) {
           metadata[key] = additionalMeta[key];
         }
       }
-      let sum = new ResultSummary(this._statement, this._parameters, metadata);
-      onCompletedOriginal.call(observer, sum);
+      const sum = new ResultSummary(this._statement, this._parameters, metadata);
+
+      // notify connection holder that the used connection is not needed any more because result has
+      // been fully consumed; call the original onCompleted callback after that
+      self._connectionHolder.releaseConnection().then(() => {
+        onCompletedOriginal.call(observer, sum);
+      });
     };
     observer.onCompleted = onCompletedWrapper;
-    observer.onError = observer.onError || ((err) => {
-      console.log("Uncaught error when processing result: " + err);
+
+    const onErrorOriginal = observer.onError || (error => {
+      console.log("Uncaught error when processing result: " + error);
     });
+
+    const onErrorWrapper = error => {
+      // notify connection holder that the used connection is not needed any more because error happened
+      // and result can't bee consumed any further; call the original onError callback after that
+      self._connectionHolder.releaseConnection().then(() => {
+        onErrorOriginal.call(observer, error);
+      });
+    };
+    observer.onError = onErrorWrapper;
+
     this._streamObserver.subscribe(observer);
   }
 }