pg-logical-replication

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