Support for logical streaming replication (#1271)

* Support for logical streaming replication

* Wrong compare expr in getLibpqConnectionString

* Simplify codes for replication parameter

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;
+  }
 
   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('/'));
 
@@ -88,6 +89,9 @@ ConnectionParameters.prototype.getLibpqConnectionString = function(cb) {
   if(this.database) {
     params.push("dbname='" + this.database + "'");
   }
+  if(this.replication) {
+    params.push("replication='" + 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);