Package Exports

pg-logical-replication

This package does not declare an exports field, so the exports above have been automatically detected and optimized by JSPM instead. If any package subpath is missing, it is recommended to post an issue to the original package (pg-logical-replication) to support the "exports" field. If that is not possible, create a JSPM override to customize the exports field for this package.

Readme

pg-logical-replication

PostgreSQL Logical Replication client for node.js

1. Install

pg-logical-replication depends on pq (node-postgres) >= 6.2.2

$ npm install pg-logical-replication

2. LogicalReplication

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*/});

3. Stream

3-1. Method - Stream.getChanges

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 properties
- includeXids : bool (default: false)
- includeTimestamp : bool (default: false)
- skipEmptyXacts : bool (default: true)

3-2. Method - Stream.stop

Stop WAL streaming.

stream.stop();

3-3. Event - Stream.on('data')

Raised when new data streamed from PostgreSQL server.

stream.on('data', (/*object*/ msg)=>{/*...*/});

msg contains lsn (string), log (buffer)

3-4. Event - Stream.on('error')

Raised when error or disconnected.

stream.on('error', (/*object*/ err)=>{/*...*/});

4. Plugin

4-1. test_decoding output

If you are using test_decoding, this plugin will be useful.

var PluginTestDecoding = LogicalReplication.LoadPlugin('output/test_decoding');
PluginTestDecoding.parse(msg.log.toString('utf8'));

Example

/*
 * 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);
        }
    });
})();

PostgreSQL side

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');