Prepared for stateless transactions

- also bugfix routing discovery query
- rename constant for semantics
- tmp bind query()
- rename prop for semantics
- README

Author: Mr-Squirrel

README.md

@@ -39,46 +39,52 @@ const driver = new Driver({});
 // Reactive
 driver.query('RETURN $foo', {foo: true}).subscribe({
     next: console.log,
-    complete: driver.shutDown,
-    error: driver.shutDown
+    complete: () => driver.shutDown().toPromise(),
+    error: (err) => {
+        console.error(err);
+        
+        driver.shutDown().toPromise();
+    }
 })
 
 // Promise
 driver.query('RETURN $foo', {foo: true})
     .toPromise()
     .then(console.log)
     .catch(console.error)
-    .finally(driver.shutDown);
+    .finally(() => driver.shutDown().toPromise());
 ```
 
 ### Transactions
 Only for 4.X
 ```Typescript
-import {flatMap, tap} from 'rxjs/operators';
+import {flatMap, reduce, tap} from 'rxjs/operators';
 
-import {Driver, Num} from '.';
+import {Driver, List, Num, Result} from '.';
 
-const driver = new Driver({});
+const driver = new Driver<Result>({});
 
 // Reactive
 driver.transaction().pipe(
     flatMap((tx) => tx.query('CREATE (n {foo: $foo}) RETURN n', {foo: true}).pipe(
-        tap(({data}) => data.length.greaterThan(Num.ZERO)
-            ? tx.commit()
-            : tx.rollback()
-        )
+        reduce((agg, next) => agg.concat(next), List.of<Result>([])),
+        tap(() => tx.rollback().toPromise())
     ))
 ).subscribe({
     next: console.log,
-    complete: driver.shutDown,
-    error: driver.shutDown
+    complete: () => driver.shutDown().toPromise(),
+    error: (err) => {
+        console.error(err);
+
+        driver.shutDown().toPromise();
+    }
 });
 
 // Promise
 getResults()
     .then(console.log)
     .catch(console.error)
-    .finally(driver.shutDown)
+    .finally(() => driver.shutDown().toPromise())
 
 async function getResults()  {
     const tx = await driver.transaction().toPromise();
@@ -103,7 +109,7 @@ import {forkJoin} from 'rxjs';
 import {filter, reduce} from 'rxjs/operators';
 import _ from 'lodash'
 
-import {Driver, DRIVER_RESULT_TYPE, List, Result} from './index';
+import {Driver, DRIVER_RESULT_TYPE, List, Result} from '.';
 
 const driver = new Driver<Result>({
     useRouting: true,
@@ -120,8 +126,12 @@ const result = forkJoin(_.map(Array(10), () => query));
 
 result.subscribe({
     next: console.log,
-    error: console.error,
-    complete: driver.shutDown
+    complete: () => driver.shutDown().toPromise(),
+    error: (err) => {
+        console.error(err);
+
+        driver.shutDown().toPromise();
+    }
 })
 
 // Promise
@@ -130,15 +140,15 @@ const result = Promise.all(_.map(Array(10), () => query.toPromise()));
 result
     .then(console.log)
     .catch(console.error)
-    .finally(driver.shutDown);
+    .finally(() => driver.shutDown().toPromise())
 ```
 
 ## Routing + Transactions
 Only for 4.X
 ```TypeScript
 import {filter, reduce} from 'rxjs/operators';
 
-import {DBMS_MEMBER_ROLE, Driver, DRIVER_RESULT_TYPE, List, Result} from './index';
+import {DBMS_DB_ROLE, Driver, DRIVER_RESULT_TYPE, List, Result} from '.';
 
 const driver = new Driver<Result>({
     useRouting: true,
@@ -148,12 +158,12 @@ const driver = new Driver<Result>({
 getResults()
     .then(console.log)
     .catch(console.error)
-    .finally(driver.shutDown)
+    .finally(() => driver.shutDown().toPromise())
 
 async function getResults()  {
     // request WRITE transaction for db 'neo4j'
-    const tx = await driver.transaction({role: DBMS_MEMBER_ROLE.LEADER, db: 'neo4j'}).toPromise();
-    const q1 = await tx.query('CREATE (n {foo: $foo})', {foo: true}).pipe(
+    const tx = await driver.transaction({role: DBMS_DB_ROLE.LEADER, db: 'neo4j'}).toPromise();
+    const q1 = await tx.query('CREATE (n {foo: $foo}) RETURN n', {foo: true}).pipe(
        filter(({type}) => type === DRIVER_RESULT_TYPE.RECORD),
        reduce((agg, next) => agg.concat(next), List.of<Result>([]))
    ).toPromise();
@@ -185,8 +195,12 @@ driver.query('MATCH (n) RETURN n')
         reduce((agg, next) => agg.concat(next), [])
     ).subscribe({
         next: console.log,
-        complete: driver.shutDown,
-        error: driver.shutDown
+        complete: () => driver.shutDown().toPromise(),
+        error: (err) => {
+            console.error(err);
+            
+            driver.shutDown().toPromise();
+        }
     })
 ```
 

src/connection/connection.class.ts

@@ -13,7 +13,7 @@ export default class Connection<Data extends any = any> extends Subject<IServerM
     public readonly id = uuid();
 
     protected readonly socket: WebSocket;
-    protected protocol: BOLT_PROTOCOLS = BOLT_PROTOCOLS.UNKNOWN;
+    protected ourProtocol: BOLT_PROTOCOLS = BOLT_PROTOCOLS.UNKNOWN;
     protected didAuth: boolean = false;
     protected incomingData = new ArrayBuffer(0);
     protected readySubject = new AsyncSubject<void>();
@@ -37,8 +37,12 @@ export default class Connection<Data extends any = any> extends Subject<IServerM
         return `${this.config.host}:${this.config.port}`;
     }
 
+    public get protocol() {
+        return this.ourProtocol;
+    }
+
     private get didHandshake() {
-        return this.protocol !== BOLT_PROTOCOLS.UNKNOWN;
+        return this.ourProtocol !== BOLT_PROTOCOLS.UNKNOWN;
     }
 
     private get isReady() {
@@ -56,13 +60,13 @@ export default class Connection<Data extends any = any> extends Subject<IServerM
             map(() => {
                 const {cmd, data, additionalData} = message;
                 const allData = additionalData
-                    ? [...data, ...additionalData(this.protocol)]
+                    ? [...data, ...additionalData(this.ourProtocol)]
                     : [...data];
 
-                this.socket.send(createMessage<Data>(this.protocol, cmd, allData, this.config.packer));
+                this.socket.send(createMessage<Data>(this.ourProtocol, cmd, allData, this.config.packer));
             }),
             switchMap(() => this)
-        )
+        );
     }
 
     @boundMethod
@@ -77,10 +81,10 @@ export default class Connection<Data extends any = any> extends Subject<IServerM
 
         return this.terminationSubject.pipe(
             map(() => {
-                this.socket.close()
+                this.socket.close();
             }),
             mapTo(this)
-        // @todo: not very graceful
+            // @todo: not very graceful
         ).toPromise();
     }
 
@@ -123,7 +127,7 @@ export default class Connection<Data extends any = any> extends Subject<IServerM
 
     @boundMethod
     private onChunk(view: DataView) {
-        const {data} = unpackResponseData<Data>(this.protocol, view, this.config.unpacker);
+        const {data} = unpackResponseData<Data>(this.ourProtocol, view, this.config.unpacker);
 
         this.next({
             // @todo: cleanup
@@ -134,9 +138,9 @@ export default class Connection<Data extends any = any> extends Subject<IServerM
 
     @boundMethod
     private onHandshake(data: DataView) {
-        this.protocol = data.getInt32(0, false);
+        this.ourProtocol = data.getInt32(0, false);
 
-        this.socket.send(getAuthMessage(this.protocol, this.config, this.config.packer));
+        this.socket.send(getAuthMessage(this.ourProtocol, this.config, this.config.packer));
     }
 
     @boundMethod

src/connection/connection.utils.ts

@@ -60,7 +60,7 @@ function getCommandForProtocol(protocol: BOLT_PROTOCOLS, cmd: DriverCommand): nu
                 return V3_BOLT_MESSAGES[cmd];
 
             default: {
-                throw new InvalidOperationError(`Protocol ${protocol} does not support transactions`)
+                throw new InvalidOperationError(`Protocol ${protocol} does not support transactions`);
             }
         }
     }
@@ -106,10 +106,10 @@ export function createMessage<T extends any = any>(protocol: BOLT_PROTOCOLS, cmd
 
 function getMessageHeader(size: number, signature: number) {
     if (size < 0x10) {
-        return [HEADER_SIZE_LIMITS.TINY_STRUCT | size, signature]
+        return [HEADER_SIZE_LIMITS.TINY_STRUCT | size, signature];
     }
 
-    throw new Error('Messages of size ' + size + ' are not supported')
+    throw new Error('Messages of size ' + size + ' are not supported');
 }
 
 export function getAuthMessage<T extends any = any>(protocol: BOLT_PROTOCOLS, params: IConnectionConfig<any>, packer?: Packer<T>) {

src/dev.ts

@@ -1,22 +1,19 @@
-import {DBMS_MEMBER_ROLE, Driver, List, Result} from '.';
+import {DBMS_DB_ROLE, Driver, List, Result} from '.';
 import {reduce} from 'rxjs/operators';
 
 const driver = new Driver<Result>({
     useRouting: true,
-    maxPoolSize: 10,
-    connectionConfig: {
-        port: 7697
-    }
+    maxPoolSize: 10
 });
 
 // Promise
 getResults()
     .then(console.log)
     .catch(console.error)
-    .finally(driver.shutDown);
+    .finally(() => driver.shutDown().toPromise());
 
 async function getResults() {
-    const tx = await driver.transaction({role: DBMS_MEMBER_ROLE.LEADER, db: 'neo4j'}).toPromise();
+    const tx = await driver.transaction({role: DBMS_DB_ROLE.LEADER, db: 'neo4j'}).toPromise();
     const q1 = await tx.query('CREATE (n {foo: $foo}) RETURN n', {foo: true}).pipe(
         reduce((agg, next) => agg.concat(next), List.of<Result>([]))
     ).toPromise();

src/driver/driver.abstract.ts

@@ -27,25 +27,29 @@ import {InvalidOperationError} from '../errors';
 import {determineConnectionHosts} from './driver.utils';
 
 export default abstract class DriverBase<Rec = any> {
+    protected isReady = false;
+    protected isShutDown = false;
+    protected isSlave = false;
     protected discoveryTables: IDiscoveryTable[] = [];
-    protected readonly connectionConfig: IConnectionConfig;
     protected connections: Connection[] = [];
     protected available: Connection[] = [];
-    protected readySubject = new AsyncSubject<void>();
-    protected availableConnections: BehaviorSubject<Connection[]> = new BehaviorSubject<Connection[]>([]);
-    protected requestQueue: Subject<IRequest> = new Subject();
-    protected processing: Observable<[IRequest, Connection]> = this.requestQueue.pipe(
+    protected readonly connectionConfig: IConnectionConfig;
+    protected readonly connectionSubject: BehaviorSubject<Connection[]> = new BehaviorSubject<Connection[]>([]);
+    protected readonly readySubject = new AsyncSubject<void>();
+    protected readonly availableConnections: BehaviorSubject<Connection[]> = new BehaviorSubject<Connection[]>([]);
+    protected readonly requestQueue: Subject<IRequest> = new Subject();
+    protected readonly processing: Observable<[IRequest, Connection]> = this.requestQueue.pipe(
         flatMap((request) => this.requestAvailableConnection(request.meta).pipe(
             map((connections): [IRequest, Connection] => [request, connections[0]])
         )),
     );
-    protected isReady = false;
-    protected isShutDown = false;
-    protected isBound = false;
 
     constructor(protected readonly config: IDriverConfig) {
         this.connectionConfig = _.merge({}, DEFAULT_CONNECTION_CONFIG, this.config.connectionConfig);
 
+        // @boundMethod does not like inheritance
+        this.query = this.query.bind(this);
+
         if (!this.config.useRouting) {
             this.isReady = true;
 
@@ -55,7 +59,7 @@ export default abstract class DriverBase<Rec = any> {
         }
     }
 
-    //@boundMethod
+    // @boundMethod does not like inheritance
     query<Res = Rec>(cypher: string, params: any = {}, meta: IQueryMeta = {}): Observable<Res> {
         const {pullN = -1, db} = meta; // @todo: transaction session ID
 
@@ -66,7 +70,7 @@ export default abstract class DriverBase<Rec = any> {
                     data: [cypher, params],
                     additionalData: (protocol) => {
                         if (protocol < BOLT_PROTOCOLS.V3) {
-                            return []
+                            return [];
                         }
 
                         return db && protocol > BOLT_PROTOCOLS.V3
@@ -90,26 +94,26 @@ export default abstract class DriverBase<Rec = any> {
         throw new InvalidOperationError('Transaction pending');
     };
 
-    commit() {
+    commit<Res = Rec>(): Observable<Res> {
         throw new InvalidOperationError('No transaction pending');
     }
 
-    rollback() {
+    rollback<Res = Rec>(): Observable<Res> {
         throw new InvalidOperationError('No transaction pending');
     }
 
-    abstract shutDown(): Promise<this>;
+    abstract shutDown(): Observable<this>;
 
     @boundMethod
     protected sendMessages<Res>(messages: IClientMessage[] = [], meta?: IRequestMeta): Observable<Res> {
         return this.readySubject.pipe(
-            flatMap(() => this.getConnectionForRequest(messages, meta)),
+            flatMap(() => this.waitForConnection(messages, meta)),
             flatMap(([request, connection]) => this.executeRequests<Res>(request, connection))
         );
     }
 
     @boundMethod
-    protected getConnectionForRequest(messages: IClientMessage[], meta?: IRequestMeta): Observable<[IRequest, Connection]> {
+    protected waitForConnection(messages: IClientMessage[], meta?: IRequestMeta): Observable<[IRequest, Connection]> {
         if (this.isShutDown) {
             throw new InvalidOperationError('Driver is shut down');
         }
@@ -130,7 +134,6 @@ export default abstract class DriverBase<Rec = any> {
             take(1),
             tap(([, connection]) => this.occupyConnection(connection)),
         );
-
     }
 
     @boundMethod
@@ -277,8 +280,8 @@ export default abstract class DriverBase<Rec = any> {
 
         const validAvailable = _.filter(this.available, connectionPredicate);
 
-        if (this.isBound || arrayHasItems(validAvailable)) {
-            return this.availableConnections
+        if (this.isSlave || arrayHasItems(validAvailable)) {
+            return this.availableConnections;
         }
 
         if (this.connections.length < this.config.maxPoolSize) {
@@ -290,19 +293,21 @@ export default abstract class DriverBase<Rec = any> {
             ];
 
             this.releaseConnection(connection);
+            this.connectionSubject.next(this.connections);
             this.availableConnections.next(this.available);
         }
 
         return this.availableConnections.pipe(
             skipWhile((connections) => !_.some(connections, connectionPredicate))
-        )
+        );
     }
 
     @boundMethod
     protected terminateConnection(connection: Connection): Promise<Connection> {
         this.occupyConnection(connection);
 
         this.connections = _.filter(this.connections, ({id}) => id !== connection.id);
+        this.connectionSubject.next(this.connections);
 
         return connection.terminate();
     }