More byte arrays stuff

Author: lutovich

src/v1/internal/connection-providers.js

@@ -54,8 +54,7 @@ export class DirectConnectionProvider extends ConnectionProvider {
   }
 
   acquireConnection(mode) {
-    const connection = this._connectionPool.acquire(this._address);
-    const connectionPromise = Promise.resolve(connection);
+    const connectionPromise = acquireConnectionFromPool(this._connectionPool, this._address);
     return this._withAdditionalOnErrorCallback(connectionPromise, this._driverOnErrorCallback);
   }
 }
@@ -102,7 +101,7 @@ export class LoadBalancer extends ConnectionProvider {
         `Failed to obtain connection towards ${serverName} server. Known routing table is: ${this._routingTable}`,
         SESSION_EXPIRED));
     }
-    return this._connectionPool.acquire(address);
+    return acquireConnectionFromPool(this._connectionPool, address);
   }
 
   _freshRoutingTable(accessMode) {
@@ -199,10 +198,9 @@ export class LoadBalancer extends ConnectionProvider {
   }
 
   _createSessionForRediscovery(routerAddress) {
-    const connection = this._connectionPool.acquire(routerAddress);
     // 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 initializedConnectionPromise = acquireConnectionFromPool(this._connectionPool, routerAddress);
     const connectionProvider = new SingleConnectionProvider(initializedConnectionPromise);
     return new Session(READ, connectionProvider);
   }
@@ -263,3 +261,19 @@ export class SingleConnectionProvider extends ConnectionProvider {
     return connectionPromise;
   }
 }
+
+// todo: test that all connection providers return initialized connections
+
+/**
+ * Acquire an initialized connection from the given connection pool for the given address. Returned connection
+ * promise will be resolved by a connection which completed initialization, i.e. received a SUCCESS response
+ * for it's INIT message.
+ * @param {Pool} connectionPool the connection pool to acquire connection from.
+ * @param {string} address the server address.
+ * @return {Promise<Connection>} the initialized connection.
+ */
+function acquireConnectionFromPool(connectionPool, address) {
+  const connection = connectionPool.acquire(address);
+  // initialized connection is required to be able to perform subsequent server version checks
+  return connection.initializationCompleted();
+}

src/v1/internal/connector.js

@@ -27,6 +27,7 @@ import {newError} from './../error';
 import ChannelConfig from './ch-config';
 import {parseHost, parsePort} from './util';
 import StreamObserver from './stream-observer';
+import {ServerVersion, VERSION_3_2_0} from './server-version';
 
 let Channel;
 if( NodeChannel.available ) {
@@ -472,8 +473,17 @@ class Connection {
       return this._packer.packable(value, (err) => this._handleFatalError(err));
   }
 
-  setServerVersion(version) {
-    this.server.version = version;
+  /**
+   * @protected
+   */
+  _markInitialized(metadata) {
+    const serverVersion = metadata.server;
+    if (!this.server.version) {
+      this.server.version = serverVersion;
+      if (ServerVersion.fromString(serverVersion).compareTo(VERSION_3_2_0) < 0) {
+        this._packer.disableByteArrays();
+      }
+    }
   }
 }
 
@@ -524,7 +534,7 @@ class ConnectionState {
       },
       onCompleted: metaData => {
         if (metaData && metaData.server) {
-          this._connection.setServerVersion(metaData.server);
+          this._connection._markInitialized(metaData);
         }
         this._initialized = true;
         if (this._resolvePromise) {

src/v1/internal/packstream.js

@@ -16,9 +16,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import utf8 from "./utf8";
-import Integer, {int, isInt} from "../integer";
-import {newError} from "./../error";
+import utf8 from './utf8';
+import Integer, {int, isInt} from '../integer';
+import {newError} from './../error';
 
 const TINY_STRING = 0x80;
 const TINY_LIST = 0x90;
@@ -38,6 +38,9 @@ const STRING_32 = 0xD2;
 const LIST_8 = 0xD4;
 const LIST_16 = 0xD5;
 const LIST_32 = 0xD6;
+const BYTES_8 = 0xCC;
+const BYTES_16 = 0xCD;
+const BYTES_32 = 0xCE;
 const MAP_8 = 0xD8;
 const MAP_16 = 0xD9;
 const MAP_32 = 0xDA;
@@ -74,6 +77,7 @@ class Structure {
 class Packer {
   constructor (channel) {
     this._ch = channel;
+    this._byteArraysSupported = true;
   }
 
   /**
@@ -95,6 +99,8 @@ class Packer {
       return () => this.packString(x, onError);
     } else if (isInt(x)) {
       return () => this.packInteger( x );
+    } else if (x instanceof Int8Array) {
+      return () => this.packBytes(x, onError);
     } else if (x instanceof Array) {
       return () => {
         this.packListHeader(x.length, onError);
@@ -225,6 +231,36 @@ class Packer {
     }
   }
 
+  packBytes(array, onError) {
+    if(this._byteArraysSupported) {
+      this.packBytesHeader(array.length, onError);
+      for (let i = 0; i < array.length; i++) {
+        this._ch.writeInt8(array[i]);
+      }
+    }else {
+      onError(newError("Byte arrays are not supported by the database this driver is connected to"));
+    }
+  }
+
+  packBytesHeader(size, onError) {
+    if (size < 0x100) {
+      this._ch.writeUInt8(BYTES_8);
+      this._ch.writeUInt8(size);
+    } else if (size < 0x10000) {
+      this._ch.writeUInt8(BYTES_16);
+      this._ch.writeUInt8((size / 256 >> 0) % 256);
+      this._ch.writeUInt8(size % 256);
+    } else if (size < 0x100000000) {
+      this._ch.writeUInt8(BYTES_32);
+      this._ch.writeUInt8((size / 16777216 >> 0) % 256);
+      this._ch.writeUInt8((size / 65536 >> 0) % 256);
+      this._ch.writeUInt8((size / 256 >> 0) % 256);
+      this._ch.writeUInt8(size % 256);
+    } else {
+      onError(newError('Byte arrays of size ' + size + ' are not supported'));
+    }
+  }
+
   packMapHeader (size, onError) {
     if (size < 0x10) {
       this._ch.writeUInt8(TINY_MAP | size);
@@ -262,6 +298,10 @@ class Packer {
       onError(newError("Structures of size " + size + " are not supported"));
     }
   }
+
+  disableByteArrays() {
+    this._byteArraysSupported = false;
+  }
 }
 
 /**
@@ -284,6 +324,14 @@ class Unpacker {
     return value;
   }
 
+  unpackBytes(size, buffer) {
+    const value = new Int8Array(size);
+    for (let i = 0; i < size; i++) {
+      value[i] = buffer.readInt8();
+    }
+    return value;
+  }
+
   unpackMap (size, buffer) {
     let value = {};
     for(let i = 0; i < size; i++) {
@@ -344,6 +392,12 @@ class Unpacker {
       return this.unpackList(buffer.readUInt16(), buffer);
     } else if (marker == LIST_32) {
       return this.unpackList(buffer.readUInt32(), buffer);
+    } else if (marker == BYTES_8) {
+      return this.unpackBytes(buffer.readUInt8(), buffer);
+    } else if (marker == BYTES_16) {
+      return this.unpackBytes(buffer.readUInt16(), buffer);
+    } else if (marker == BYTES_32) {
+      return this.unpackBytes(buffer.readUInt32(), buffer);
     } else if (marker == MAP_8) {
       return this.unpackMap(buffer.readUInt8(), buffer);
     } else if (marker == MAP_16) {

test/internal/shared-neo4j.js

@@ -95,7 +95,7 @@ const password = 'password';
 const authToken = neo4j.auth.basic(username, password);
 
 const neoCtrlVersionParam = '-e';
-const defaultNeo4jVersion = '3.1.3';
+const defaultNeo4jVersion = '3.2.0';
 const defaultNeoCtrlArgs = `${neoCtrlVersionParam} ${defaultNeo4jVersion}`;
 
 function neo4jCertPath(dir) {

test/v1/types.test.js

@@ -19,6 +19,8 @@
 
 import neo4j from '../../src/v1';
 import sharedNeo4j from '../internal/shared-neo4j';
+import _ from 'lodash';
+import {ServerVersion, VERSION_3_2_0} from '../../src/v1/internal/server-version';
 
 describe('floating point values', () => {
   it('should support float 1.0 ', testValue(1));
@@ -136,18 +138,117 @@ describe('path values', () => {
   });
 });
 
-function testValue(actual, expected) {
-  return done => {
+describe('byte arrays', () => {
+
+  let originalTimeout = jasmine.DEFAULT_TIMEOUT_INTERVAL;
+  let serverSupportsByteArrays = false;
+
+  beforeEach(done => {
+    jasmine.DEFAULT_TIMEOUT_INTERVAL = 60000;
+
     const driver = neo4j.driver('bolt://localhost', sharedNeo4j.authToken);
     const session = driver.session();
+    session.run('RETURN 1').then(result => {
+      driver.close();
+      const serverVersion = ServerVersion.fromString(result.summary.server.version);
+      serverSupportsByteArrays = serverVersion.compareTo(VERSION_3_2_0) >= 0;
+      done();
+    });
+  });
 
-    session.run('RETURN {val} as v', {val: actual})
-      .then(result => {
-        expect(result.records[0].get('v')).toEqual(expected || actual);
-        driver.close();
-        done();
-      }).catch(err => {
-      console.log(err);
+  afterEach(() => {
+    jasmine.DEFAULT_TIMEOUT_INTERVAL = originalTimeout;
+  });
+
+  it('should support returning empty byte array', done => {
+    if(!serverSupportsByteArrays) {
+      done();
+      return;
+    }
+
+    testValue(new Int8Array(0))(done);
+  });
+
+  it('should support returning empty byte array', conditionalTestValues(serverSupportsByteArrays, new Int8Array(0)));
+
+  it('should support returning short byte arrays', conditionalTestValues(serverSupportsByteArrays, randomByteArrays(100, 1, 255)));
+
+  it('should support returning medium byte arrays', conditionalTestValues(serverSupportsByteArrays, randomByteArrays(50, 256, 65535)));
+
+  it('should support returning long byte arrays', conditionalTestValues(serverSupportsByteArrays, randomByteArrays(10, 65536, 2 * 65536)));
+
+  it('should fail to return byte array', done => {
+    if (serverSupportsByteArrays) {
+      done();
+      return;
+    }
+
+    const driver = neo4j.driver('bolt://localhost', sharedNeo4j.authToken);
+    const session = driver.session();
+    session.run('RETURN {array}', {array: randomByteArray(42)}).catch(error => {
+      driver.close();
+      expect(error.message).toEqual('Byte arrays are not supported by the database this driver is connected to');
+      done();
+    });
+  });
+});
+
+function conditionalTestValues(condition, values) {
+  if (!condition) {
+    return done => done();
+  }
+
+  const driver = neo4j.driver('bolt://localhost', sharedNeo4j.authToken);
+  const queriesPromise = values.reduce((acc, value) =>
+    acc.then(() => runReturnQuery(driver, value)), Promise.resolve());
+  return asTestFunction(queriesPromise, driver);
+}
+
+function testValue(actual, expected) {
+  const driver = neo4j.driver('bolt://localhost', sharedNeo4j.authToken);
+  const queryPromise = runReturnQuery(driver, actual, expected);
+  return asTestFunction(queryPromise, driver);
+}
+
+function testValues(values) {
+  const driver = neo4j.driver('bolt://localhost', sharedNeo4j.authToken);
+  const queriesPromise = values.reduce((acc, value) =>
+    acc.then(() => runReturnQuery(driver, value)), Promise.resolve());
+  return asTestFunction(queriesPromise, driver);
+}
+
+function runReturnQuery(driver, actual, expected) {
+  const session = driver.session();
+  return new Promise((resolve, reject) => {
+    session.run('RETURN {val} as v', {val: actual}).then(result => {
+      expect(result.records[0].get('v')).toEqual(expected || actual);
+      session.close();
+      resolve();
+    }).catch(error => {
+      reject(error);
     });
-  };
+  });
+}
+
+function asTestFunction(promise, driver) {
+  return done =>
+    promise.then(() => {
+      driver.close();
+      done();
+    }).catch(error => {
+      driver.close();
+      console.log(error);
+    });
+}
+
+function randomByteArrays(count, minLength, maxLength) {
+  return _.range(count).map(() => {
+    const length = _.random(minLength, maxLength);
+    return randomByteArray(length);
+  });
+}
+
+function randomByteArray(length) {
+  const array = _.range(length).map(() => _.random(-128, 127));
+  return new Int8Array(array);
 }