Plug new pool into driver

Author: jakewins

src/v1/driver.js

@@ -18,6 +18,7 @@
  */
 
 import Session from './session';
+import {Pool} from './internal/pool';
 import {connect} from "./internal/connector";
 
 /**
@@ -37,24 +38,74 @@ class Driver {
     this._openSessions = {};
     this._sessionIdGenerator = 0;
     this._token = token || {};
+    this._pool = new Pool(
+      this._createConnection.bind(this),
+      this._destroyConnection.bind(this),
+      this._validateConnection.bind(this)
+    );
   }
 
   /**
-   * Create and return new session
-   * @return {Session} new session.
+   * Create a new connection instance.
+   * @return {Connection} new connector-api session instance, a low level session API.
+   * @access private
    */
-  session() {
+  _createConnection( release ) {
     let sessionId = this._sessionIdGenerator++;
     let conn = connect(this._url);
     conn.initialize(this._userAgent, this._token);
-    let _driver = this;
-    let _session = new Session( conn, () => {
-      // On close of session, remove it from the list of open sessions
-      delete _driver._openSessions[sessionId];
-    });
+    conn._id = sessionId;
+    conn._release = () => release(conn);
+
+    this._openSessions[sessionId] = conn;
+    return conn;
+  }
+
+  /**
+   * Check that a connection is usable
+   * @return {boolean} true if the connection is open
+   * @access private
+   **/
+  _validateConnection( conn ) {
+    return conn.isOpen();
+  }
+
+  /**
+   * Dispose of a live session, closing any associated resources.
+   * @return {Session} new session.
+   * @access private
+   */
+  _destroyConnection( conn ) {
+    delete this._openSessions[conn._id];
+    conn.close();
+  }
+
+  /**
+   * Create and return new session
+   * @return {Session} new session.
+   */
+  session() {
+    let conn = this._pool.acquire();
+    return new Session( conn, (cb) => {
+      // This gets called on Session#close(), and is where we return
+      // the pooled 'connection' instance.
+
+      // 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.
 
-    this._openSessions[sessionId] = _session;
-    return _session;
+      // Queue up a 'reset', to ensure the next user gets a clean
+      // session to work with. No need to flush, this will get sent
+      // along with whatever the next thing the user wants to do with
+      // this session ends up being, so we save the network round trip.
+      conn.reset();
+
+      // Return connection to the pool
+      conn._release();
+
+      // Call user callback
+      if(cb) { cb(); }
+    });
   }
 
   /**

src/v1/internal/ch-node.js

@@ -60,6 +60,12 @@ class NodeChannel {
         _self.onmessage( new NodeBuffer( buffer ) );
       }
     });
+
+    this._conn.on('error', function(err){
+      if( _self.onerror ) {
+        _self.onerror(err);
+      }
+    });
   }
 
   /**

src/v1/internal/ch-websocket.js

@@ -54,6 +54,12 @@ class WebSocketChannel {
         self.onmessage( b );
       } 
     };
+
+    this._ws.onerror = () => {
+      if( self.onerror ) {
+        self.onerror();
+      }
+    }
   }
 
   /**

src/v1/internal/connector.js

@@ -40,7 +40,8 @@ else {
 let
 // Signature bytes for each message type
 INIT = 0x01,            // 0000 0001 // INIT <user_agent>
-ACK_FAILURE = 0x0F,     // 0000 1111 // ACK_FAILURE
+ACK_FAILURE = 0x0D,     // 0000 1101 // ACK_FAILURE
+RESET = 0x0F,           // 0000 1111 // RESET
 RUN = 0x10,             // 0001 0000 // RUN <statement> <parameters>
 DISCARD_ALL = 0x2F,     // 0010 1111 // DISCARD *
 PULL_ALL = 0x3F,        // 0011 1111 // PULL *
@@ -176,6 +177,9 @@ class Connection {
     this._unpacker = new packstream.Unpacker();
     this._isHandlingFailure = false;
 
+    // Set to true on fatal errors, to get this out of session pool.
+    this._isBroken = false;
+
     // For deserialization, explain to the unpacker how to unpack nodes, rels, paths;
     this._unpacker.structMappers[NODE] = _mappers.node;
     this._unpacker.structMappers[RELATIONSHIP] = _mappers.rel;
@@ -197,14 +201,16 @@ class Connection {
         }
 
       } else {
-        // TODO: Report error
+        this._isBroken = true;
         console.log("FATAL, unknown protocol version:", proposed)
       }
     };
 
+    this._ch.onerror = () => self._isBroken = true;
+
     this._dechunker.onmessage = (buf) => {
       self._handleMessage( self._unpacker.unpack( buf ) );
-    }
+    };
 
     let handshake = alloc( 5 * 4 );
     //magic preamble
@@ -268,6 +274,7 @@ class Connection {
         }
         break;
       default:
+        this._isBroken = true;
         console.log("UNKNOWN MESSAGE: ", msg);
     }
   }
@@ -300,6 +307,13 @@ class Connection {
     this._chunker.messageBoundary();
   }
 
+  /** Queue a RESET-message to be sent to the database */
+  reset( observer ) {
+    this._queueObserver(observer);
+    this._packer.packStruct( RESET );
+    this._chunker.messageBoundary();
+  }
+
   /** Queue a ACK_FAILURE-message to be sent to the database */
   _ackFailure( observer ) {
     this._queueObserver(observer);
@@ -324,6 +338,11 @@ class Connection {
     this._chunker.flush();
   }
 
+  /** Check if this connection is in working condition */
+  isOpen() {
+    return !this._isBroken && this._ch._open;
+  }
+
   /**
    * Call close on the channel.
    * @param {function} cb - Function to call on close.

src/v1/session.js

@@ -47,14 +47,14 @@ class Session {
    * @param {Object} parameters - Map with parameters to use in statement
    * @return {Result} - New Result
    */
-  run(statement, parameters) {
+  run(statement, parameters = {}) {
     if(typeof statement === 'object' && statement.text) {
       parameters = statement.parameters || {};
       statement = statement.text;
     }
     let streamObserver = new StreamObserver();
     if (!this._hasTx) {
-      this._conn.run(statement, parameters || {}, streamObserver);
+      this._conn.run(statement, parameters, streamObserver);
       this._conn.pullAll(streamObserver);
       this._conn.sync();
     } else {
@@ -86,9 +86,8 @@ class Session {
    * @param {function()} cb - Function to be called after the session has been closed
    * @return
    */
-  close(cb) {
-    this._onClose();
-    this._conn.close(cb);
+  close(cb=(()=>null)) {
+    this._onClose(cb);
   }
 }
 

test/internal/pool.test.js

@@ -19,7 +19,7 @@
 
 var Pool = require('../../lib/v1/internal/pool').Pool;
 
-fdescribe('Pool', function() {
+describe('Pool', function() {
   it('allocates if pool is empty', function() {
     // Given
     var counter = 0;

test/v1/examples.test.js

@@ -21,7 +21,7 @@ var neo4j = require("../../lib/v1");
 
 var _console = console;
 
-describe('transaction', function() {
+describe('examples', function() {
 
   var driver, session, out, console;