Merge with 1.0

Author: RagnarW

src/v1/driver.js

@@ -18,7 +18,9 @@
  */
 
 import Session from './session';
+import {Pool} from './internal/pool';
 import {connect} from "./internal/connector";
+import StreamObserver from './internal/stream-observer';
 
 /**
   * A Driver instance is used for mananging {@link Session}s.
@@ -37,24 +39,75 @@ 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 a new connection instance.
+   * @return {Connection} new connector-api session instance, a low level session API.
+   * @access private
+   */
+  _createConnection( release ) {
+    let sessionId = this._sessionIdGenerator++;
+    let streamObserver = new _ConnectionStreamObserver(this);
+    let conn = connect(this._url);
+    conn.initialize(this._userAgent, this._token, streamObserver);
+    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 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];
-    });
+    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.
+
+      // 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();
 
-    this._openSessions[sessionId] = _session;
-    return _session;
+      // Call user callback
+      if(cb) { cb(); }
+    });
   }
 
   /**
@@ -70,4 +123,22 @@ class Driver {
   }
 }
 
+/** Internal stream observer used for connection state */
+class _ConnectionStreamObserver extends StreamObserver {
+  constructor(driver) {
+    super();
+    this._driver = driver;
+    this._hasFailed = false;
+  }
+  onError(error) {
+    if (!this._hasFailed) {
+      super.onError(error);
+      if(this._driver.onError) {
+        this._driver.onError(error);
+      }
+      this._hasFailed = true;
+    }
+  }
+}
+
 export default Driver

src/v1/graph-types.js

@@ -146,7 +146,7 @@ class PathSegment {
    * @param {Relationship} rel - Relationship segment
    * @param {string} end - Identity of end Node
    */
-  constructor( start, rel, end ) {
+  constructor(start, rel, end) {
     this.start = start;
     this.relationship = rel;
     this.end = end;
@@ -159,12 +159,14 @@ class PathSegment {
 class Path {
   /**
    * @constructor
+   * @param {Node} start  - start node
+   * @param {Node} end - end node
    * @param {Array} segments - Array of Segments
    */
-  constructor(segments) {
+  constructor(start, end, segments) {
+    this.start = start;
+    this.end = end;
     this.segments = segments;
-    this.start = segments[0].start;
-    this.end = segments[segments.length - 1].end;
     this.length = segments.length;
   }
 }

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 = (err) => {
+      if( self.onerror ) {
+        self.onerror(err);
+      }
+    }
   }
 
   /**

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 *
@@ -80,7 +81,7 @@ function LOG(err) {
 let NO_OP_OBSERVER = {
   onNext : NO_OP,
   onCompleted : NO_OP,
-  onError : LOG
+  onError : NO_OP
 }
 
 /** Maps from packstream structures to Neo4j domain objects */
@@ -114,6 +115,7 @@ let _mappers = {
         sequence = unpacker.unpack(buf);
     let prevNode = nodes[0],
         segments = [];
+
     for (let i = 0; i < sequence.length; i += 2) {
         let relIndex = sequence[i],
             nextNode = nodes[sequence[i + 1]],
@@ -138,9 +140,9 @@ let _mappers = {
         segments.push( new GraphType.PathSegment( prevNode, rel, nextNode ) );
         prevNode = nextNode;
     }
-    return new GraphType.Path( segments );
+    return new GraphType.Path(nodes[0], nodes[nodes.length - 1],  segments );
   }
-}
+};
 
 /**
  * A connection manages sending and recieving messages over a channel. A
@@ -176,6 +178,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 +202,21 @@ class Connection {
         }
 
       } else {
-        // TODO: Report error
+        this._isBroken = true;
         console.log("FATAL, unknown protocol version:", proposed)
       }
     };
 
+    this._ch.onerror = (error) => {
+      self._isBroken = true;
+      if(this._currentObserver.onError) {
+        this._currentObserver.onError(error);
+      }
+    }
+
     this._dechunker.onmessage = (buf) => {
       self._handleMessage( self._unpacker.unpack( buf ) );
-    }
+    };
 
     let handshake = alloc( 5 * 4 );
     //magic preamble
@@ -259,15 +271,16 @@ class Connection {
         break;
       case IGNORED:
         try {
-          if (this._errorMsg)
+          if (this._errorMsg && this._currentObserver.onError)
             this._currentObserver.onError(this._errorMsg);
-          else
+          else if(this._currentObserver.onError)
             this._currentObserver.onError(msg);
         } finally {
           this._currentObserver = this._pendingObservers.shift();
         }
         break;
       default:
+        this._isBroken = true;
         console.log("UNKNOWN MESSAGE: ", msg);
     }
   }
@@ -300,6 +313,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 +344,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/internal/pool.js

@@ -0,0 +1,61 @@
+/**
+ * Copyright (c) 2002-2016 "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.
+ */
+
+class Pool {
+  /**
+   * @param create  an allocation function that creates a new resource. It's given
+   *                a single argument, a function that will return the resource to
+   *                the pool if invoked, which is meant to be called on .dispose
+   *                or .close or whatever mechanism the resource uses to finalize.
+   * @param destroy called with the resource when it is evicted from this pool
+   * @param validate called at various times (like when an instance is acquired and
+   *                 when it is returned). If this returns false, the resource will
+   *                 be evicted
+   * @param maxIdle the max number of resources that are allowed idle in the pool at
+   *                any time. If this threshold is exceeded, resources will be evicted.
+   */
+  constructor(create, destroy=(()=>true), validate=(()=>true), maxIdle=50) {
+    this._create = create;
+    this._destroy = destroy;
+    this._validate = validate;
+    this._maxIdle = maxIdle;
+    this._pool = [];
+    this._release = this._release.bind(this);
+  }
+
+  acquire() {
+    if( this._pool.length > 0 ) {
+      return this._pool.pop();
+    } else {
+      return this._create( this._release );
+    }
+  }
+
+  _release(resource) {
+    if( this._pool.length >= this._maxIdle || !this._validate(resource) ) {
+      this._destroy(resource);
+    } else {
+      this._pool.push(resource);
+    }
+  }
+}
+
+export default {
+    Pool
+}

src/v1/result-summary.js

@@ -89,7 +89,7 @@ class ProfiledPlan {
   /**
    * Create a ProfiledPlan instance
    * @constructor
-   * @param {Object} plan - Object with plan data
+   * @param {Object} profile - Object with profile data
    */
   constructor(profile) {
     this.operatorType = profile.operatorType;
@@ -234,10 +234,11 @@ class Notification {
     this.code = notification.code;
     this.title = notification.title;
     this.description = notification.description;
-    this.position = this._constructPosition(notification.position);
+    this.severity = notification.severity;
+    this.position = Notification._constructPosition(notification.position);
   }
 
-  _constructPosition(pos) {
+  static _constructPosition(pos) {
     if(!pos) {
       return {};
     }
@@ -254,7 +255,7 @@ const statementType = {
   READ_WRITE: 'rw',
   WRITE_ONLY: 'w',
   SCHEMA_WRITE: 's'
-}
+};
 
 export default {
   ResultSummary,