const neo4j = require('neo4j-driver');
const yargs = require('yargs/yargs');
const {hideBin} = require('yargs/helpers');
const argv = yargs(hideBin(process.argv)).argv;
const util = require('util');
function applyChange(change) { (1)
console.log(util.inspect(change.toObject(), false, null, true));
}
async function queryChanges(driver, database, cursor, selectors) {
const session = driver.session({database: database});
try {
const current = currentChangeId(driver, database);
await session.executeRead((tx) => {
return tx.run('CALL db.cdc.query($from, $selectors)',
{from: cursor, selectors: selectors}); (2)
})
.then((res) => {
if (res.records.length === 0) {
cursor = current; (3)
} else {
res.records.map((change) => {
applyChange(change); (4)
cursor = change.get('id'); (5)
});
}
});
} catch (e) {
console.log('Failed to apply change', e);
} finally {
session.close();
}
return cursor;
}
function queryChangesLoop(driver, database, cursor, selectors) {
setTimeout(async () => {
cursor = await queryChanges(driver, database, cursor, selectors);
queryChangesLoop(driver, database, cursor, selectors);
}, 500);
}
function earliestChangeId(driver, database) { (6)
return driver.executeQuery('CALL db.cdc.earliest', {}, {database: database})
.then((res) => {
return res.records[0].get('id');
});
}
function currentChangeId(driver, database) { (7)
return driver.executeQuery('CALL db.cdc.current', {}, {'database': database})
.then((res) => {
return res.records[0].get('id');
});
}
async function main() {
const uri = argv.address ?? argv.a ?? 'neo4j://localhost:7687';
const database = argv.database ?? argv.d ?? 'neo4j';
const user = argv.user ?? argv.u ?? 'neo4j';
const password = argv.password ?? argv.p ?? 'passw0rd';
let cursor = argv.from ?? argv.f;
const driver = neo4j.driver(uri, neo4j.auth.basic(user, password));
if (!cursor) {
cursor = await currentChangeId(driver, database);
}
const selectors = [ (8)
// {"select":"n"}
];
queryChangesLoop(driver, database, cursor, selectors); (9)
// await driver.close() (10)
}
main();