Support for logical streaming replication

lib/client.js

@@ -26,6 +26,7 @@ var Client = function(config) {
   this.port = this.connectionParameters.port;
   this.host = this.connectionParameters.host;
   this.password = this.connectionParameters.password;
+  this.replication = this.connectionParameters.replication;
 
   var c = config || {};
 
@@ -222,6 +223,9 @@ Client.prototype.getStartupConf = function() {
   if (appName) {
     data.application_name = appName;
   }
+  if (params.replication) {
+    data.replication = params.replication === true ? 'true' : params.replication;
+  }
 
   return data;
 };

lib/connection-parameters.js

@@ -57,6 +57,7 @@ var ConnectionParameters = function(config) {
   this.binary = val('binary', config);
   this.ssl = typeof config.ssl === 'undefined' ? useSsl() : config.ssl;
   this.client_encoding = val("client_encoding", config);
+  this.replication = val("replication", config);
   //a domain socket begins with '/'
   this.isDomainSocket = (!(this.host||'').indexOf('/'));
 
@@ -82,6 +83,9 @@ ConnectionParameters.prototype.getLibpqConnectionString = function(cb) {
   if(this.database) {
     params.push("dbname='" + this.database + "'");
   }
+  if(this.replication) {
+    params.push("replication='" + (this.database === true ? "true" : this.replication) + "'");
+  }
   if(this.host) {
     params.push("host=" + this.host);
   }

lib/connection.js

@@ -425,6 +425,9 @@ Connection.prototype.parseMessage =  function(buffer) {
   case 0x48: //H
     return this.parseH(buffer, length);
 
+  case 0x57: //W
+    return new Message('replicationStart', length);
+
   case 0x63: //c
     return new Message('copyDone', length);
 

lib/index.js

@@ -13,6 +13,7 @@ var defaults =  require('./defaults');
 var Connection = require('./connection');
 var ConnectionParameters = require('./connection-parameters');
 var poolFactory = require('./pool-factory');
+var WalStream = require('./wallstream');
 
 var PG = function(clientConstructor) {
   EventEmitter.call(this);
@@ -23,6 +24,7 @@ var PG = function(clientConstructor) {
   this._pools = [];
   this.Connection = Connection;
   this.types = require('pg-types');
+  this.WalStream = WalStream;
 };
 
 util.inherits(PG, EventEmitter);

lib/wallstream.js

@@ -0,0 +1,92 @@
+/**
+ * Copyright (c) 2017 Kibae Shin ([email protected])
+ * All rights reserved.
+ *
+ * This source code is licensed under the MIT license found in the
+ * README.md file in the root directory of this source tree.
+ */
+var EventEmitter = require('events').EventEmitter;
+var Client = require('./client');
+var util = require('util');
+
+var WalStream = function(config) {
+  EventEmitter.call(this);
+
+  var c = config || {};
+  c.replication = 'database';
+
+  var self = this;
+
+  var client;
+  var stoped = false;
+  this.getChanges = function(slotName, uptoLsn, option, cb /*(err)*/) {
+    option = option || {};
+    /*
+     * includeXids : include xid on BEGIN and COMMIT, default false
+     * includeTimestamp : include timestamp on COMMIT, default false
+     * skipEmptyXacts : skip empty transaction like DDL, default true
+     */
+
+    stoped = false;
+    client = new Client(config);
+
+    client.on('error', function(err) {
+      self.emit('error', err);
+    });
+
+    client.connect(function(err) {
+      //error handling
+      if (err) {
+        self.emit('error', err);
+        return;
+      }
+
+      var sql = 'START_REPLICATION SLOT ' + slotName + ' LOGICAL ' + (uptoLsn ? uptoLsn : '0/00000000');
+      var opts = [
+        '"include-xids" \'' + (option.includeXids === true ? 'on' : 'off') + '\'',
+        '"include-timestamp" \'' + (option.includeTimestamp === true ? 'on' : 'off') + '\'',
+        '"skip-empty-xacts" \'' + (option.skipEmptyXacts !== false ? 'on' : 'off') + '\'',
+      ];
+      sql += ' (' + (opts.join(' , ')) + ')';
+
+      client.query(sql, function(err) {
+        if (err) {
+          if (!stoped && cb) {
+            cb(err);
+            cb = null;
+          }
+        }
+        cb = null;
+      });
+
+      client.connection.once('replicationStart', function() {
+        //start
+        self.emit('start', self);
+        client.connection.on('copyData', function(msg) {
+          if (msg.chunk[0] != 0x77) {
+            return;
+          }
+
+          var lsn = (msg.chunk.readUInt32BE(1).toString(16).toUpperCase()) + '/' + (msg.chunk.readUInt32BE(5).toString(16).toUpperCase());
+          self.emit('data', {
+            lsn: lsn,
+            log: msg.chunk.slice(25),
+          });
+        });
+      });
+    });
+    return self;
+  };
+
+  this.stop = function() {
+    stoped = true;
+    if (client) {
+      client.end();
+      client = null;
+    }
+  };
+};
+
+util.inherits(WalStream, EventEmitter);
+
+module.exports = WalStream;

test/test-walstream.js

@@ -0,0 +1,34 @@
+'use strict';
+
+const WalStream = require(__dirname + '/../lib').WalStream;
+
+var lastLsn = null;
+
+var walStream = new WalStream({});
+function proc() {
+  walStream.getChanges('test_slot', lastLsn, {
+    includeXids: false, //default: false
+    includeTimestamp: false, //default: false
+    skipEmptyXacts: true, //default: true
+  }, function(err) {
+    if (err) {
+      console.log('Logical replication initialize error', err);
+      setTimeout(proc, 1000);
+    }
+  });
+}
+
+walStream.on('data', function(msg) {
+  lastLsn = msg.lsn || lastLsn;
+
+  //DO SOMETHING. eg) replicate to other dbms(pgsql, mysql, ...)
+  console.log('log recv', msg);
+}).on('error', function(err) {
+  console.log('Error #2', err);
+  setTimeout(proc, 1000);
+});
+
+proc();
+
+//If want to stop replication
+//walStream.stop();