neo4j-devtools / tapestry

Tapestry

A neo4j driver spike illustrating an RxJS based monadic driver with full typescript support and a pluggable packstream.

Help keep the dream alive!

ToC

1. Basic usage
   • Initialisation
   • Queries
   • Transactions
   • Routing
   • Routing + Transactions
2. Custom unpackers
3. Configuration

Basic Usage

Initialisation

import {Driver} from '.';

const driver = new Driver({
    connectionConfig: {
        authToken: {
            scheme: 'basic',
            principal: 'neo4j',
            credentials: 'neo4j'
        }
    }
});

Queries

import {Driver} from '.';

const driver = new Driver({});

// Reactive
driver.query('RETURN $foo', {foo: true}).subscribe({
    next: console.log,
    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().toPromise());

Transactions

Only for 4.X

import {flatMap, reduce, tap} from 'rxjs/operators';

import {Driver, List, Num, Result} from '.';

const driver = new Driver<Result>({});

// Reactive
driver.transaction().pipe(
    flatMap((tx) => tx.query('CREATE (n {foo: $foo}) RETURN n', {foo: true}).pipe(
        reduce((agg, next) => agg.concat(next), List.of<Result>([])),
        tap(() => tx.rollback().toPromise())
    ))
).subscribe({
    next: console.log,
    complete: () => driver.shutDown().toPromise(),
    error: (err) => {
        console.error(err);

        driver.shutDown().toPromise();
    }
});

// Promise
getResults()
    .then(console.log)
    .catch(console.error)
    .finally(() => driver.shutDown().toPromise())

async function getResults()  {
    const tx = await driver.transaction().toPromise();
    const q1 = await tx.query('CREATE (n {foo: $foo}) RETURN n', {foo: true}).toPromise();

    if (q1.data.length.equals(Num.ZERO)) {
        await tx.rollback().toPromise();

        return;
    }

    await tx.commit().toPromise();

    return q1;
}

Routing

Only for 4.X

import {forkJoin} from 'rxjs';
import {filter, reduce} from 'rxjs/operators';
import _ from 'lodash'

import {Driver, DRIVER_RESULT_TYPE, List, Result} from '.';

const driver = new Driver<Result>({
    useRouting: true,
    maxPoolSize: 10
});

const query = driver.query('RETURN 1', {}).pipe(
    filter(({type}) => type === DRIVER_RESULT_TYPE.RECORD),
    reduce((agg, next) => agg.concat(next), List.of<Result>([]))
);

// Reactive
const result = forkJoin(_.map(Array(10), () => query));

result.subscribe({
    next: console.log,
    complete: () => driver.shutDown().toPromise(),
    error: (err) => {
        console.error(err);

        driver.shutDown().toPromise();
    }
})

// Promise
const result = Promise.all(_.map(Array(10), () => query.toPromise()));

result
    .then(console.log)
    .catch(console.error)
    .finally(() => driver.shutDown().toPromise())

Routing + Transactions

Only for 4.X

import {filter, reduce} from 'rxjs/operators';

import {DBMS_DB_ROLE, Driver, DRIVER_RESULT_TYPE, List, Result} from '.';

const driver = new Driver<Result>({
    useRouting: true,
    maxPoolSize: 10
});

getResults()
    .then(console.log)
    .catch(console.error)
    .finally(() => driver.shutDown().toPromise())

async function getResults()  {
    // request WRITE transaction for db 'neo4j'
    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();

    await tx.commit().toPromise();

    return q1;
}

Custom unpackers

Example using a custom JSON unpacker, removing all monads from results.

import {reduce} from 'rxjs/operators';

import {Driver, DRIVER_HEADERS, JsonUnpacker} from '.';

const driver = new Driver<any>({
    connectionConfig: {
        unpacker: JsonUnpacker,
        getResponseHeader: (data): DRIVER_HEADERS => data[0] || DRIVER_HEADERS.FAILURE,
        getResponseData: (data): any => data[1] || []
    },
    mapToResult: (headerRecord, type, data) => ({header: headerRecord, type, data})
});

driver.query('MATCH (n) RETURN n')
    .pipe(
        reduce((agg, next) => agg.concat(next), [])
    ).subscribe({
        next: console.log,
        complete: () => driver.shutDown().toPromise(),
        error: (err) => {
            console.error(err);
            
            driver.shutDown().toPromise();
        }
    })

Configuration

import {
    Packer,
    Unpacker,
    DRIVER_HEADERS,
    DRIVER_RESULT_TYPE
} from '.';

export interface IAuthToken {
    scheme: 'basic',
    principal: string,
    credentials: string;
}

export interface IConnectionConfig<Data = any> {
    secure?: true;
    authToken: IAuthToken;
    host: string;
    port: number;
    userAgent: string;
    getResponseHeader?: (unpacked: Data) => DRIVER_HEADERS,
    getResponseData?: (unpacked: Data) => Data,
    packer?: Packer<Data>;
    unpacker?: Unpacker<Data>;
}

export interface IDriverConfig<Rec = any> {
    maxPoolSize: number;
    discoveryIntervalMs: number;
    useRouting?: boolean;
    connectionConfig: Partial<IConnectionConfig>; // @todo: Partial is not correct
    mapToResultHeader: (headerRecord: any) => any;
    mapToResult: (headerRecord: any, type: DRIVER_RESULT_TYPE, data: any) => Rec;
}

tapestry's Issues

Packstream inconsistencies

There are still some types (e.g. Date time) that are not unpacked correctly. See if you can make it work

Investigate performance bottlenecks

It seems that we are incurring a penalty in rxjs if nothing else...

Support all bolt protocol versions

Currently the API only supports (ish) V1 and V2 of bolt. The packstream is still not unpacking all datatypes correctly and it does not know how to authenticate and run queries for V3 and above

Cleanup creative spike code

There are several places in the code where we have @todo: comments. Address them