- PostgreSQL Logical Replication client for node.js
- pg-logical-replication depends on pq (node-postgres) >= 6.2.2
$ npm install pg-logical-replication
new LogicalReplication( object config ) : Stream
- Creates a new, unconnected instance of a logical replication stream configured via supplied configuration object.
- https://github.com/brianc/node-postgres/wiki/Client#parameters
var LogicalReplication = require('pg-logical-replication');
var stream = new LogicalReplication({/*config*/});
- Start WAL streaming of changes.
stream.getChanges( /*string*/ slotName, /*string*/ uptoLsn, /*object*/ option, /*function(err)*/ initialErrorCallback );
uptoLsn
can be null, the minimum value is "0/00000000".option
can contain any of the following optional propertiesincludeXids
: bool (default: false)includeTimestamp
: bool (default: false)skipEmptyXacts
: bool (default: true)
- Stop WAL streaming.
stream.stop();
- Raised when new data streamed from PostgreSQL server.
stream.on('data', (/*object*/ msg)=>{/*...*/});
msg
containslsn (string)
,log (buffer)
- Raised when error or disconnected.
stream.on('error', (/*object*/ err)=>{/*...*/});
- If you are using
test_decoding
, this plugin will be useful.
var PluginTestDecoding = LogicalReplication.LoadPlugin('output/test_decoding');
PluginTestDecoding.parse(msg.log.toString('utf8'));
/*
* Test progress
* 1. Create logical replication slot with test_decoding output plugin : SELECT * FROM pg_create_logical_replication_slot('test_slot', 'test_decoding');
* 2. Launch nodejs with this file : node test.js
* 3. Modify data of database
*/
var LogicalReplication = require('pg-logical-replication');
var PluginTestDecoding = LogicalReplication.LoadPlugin('output/test_decoding');
//Connection parameter : https://github.com/brianc/node-postgres/wiki/Client#parameters
var connInfo = {};
//Initialize with last LSN value
var lastLsn = null;
var stream = (new LogicalReplication(connInfo))
.on('data', function(msg) {
lastLsn = msg.lsn || lastLsn;
var log = (msg.log || '').toString('utf8');
try {
console.log(PluginTestDecoding.parse(log));
//TODO: DO SOMETHING. eg) replicate to other dbms(pgsql, mysql, ...)
} catch (e) {
console.trace(log, e);
}
}).on('error', function(err) {
console.trace('Error #2', err);
setTimeout(proc, 1000);
});
(function proc() {
stream.getChanges('test_slot', lastLsn, {
includeXids: false, //default: false
includeTimestamp: false, //default: false
skipEmptyXacts: true, //default: true
}, function(err) {
if (err) {
console.trace('Logical replication initialize error', err);
setTimeout(proc, 1000);
}
});
})();
- Create logical replication slot
SELECT * FROM pg_create_logical_replication_slot('test_slot', 'test_decoding');
- Delete logical replication slot
SELECT pg_drop_replication_slot('test_slot');