Support for logical streaming replication

[kibae]

Hello,

Please check this code.

1. database connection for replication mode
2. WalStream class includes data event for logical replication stream
3. includes sample code test/test-walstream.js

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

[charmander]

Is this.database === true intended to read this.replication?

[kibae]

I implemented pg_recvlogical's access method the same way.
The replication value can have a value of true or database.

https://github.com/pipelinedb/pipelinedb/blob/bdcd1fcd37584a3ad5427bb1a1b27a1af7521f0c/src/bin/pg_basebackup/streamutil.c#L105

[kibae]

Oh, sorry. That's pipelinedb link..
please check this out :)

https://github.com/postgres/postgres/blob/839da3c7d9b76b178910ce2999f7cbbdf6e91f70/src/bin/pg_basebackup/streamutil.c#L110

[charmander]

So it’s intended to have:

{
    database: true,
    replication: 'truthy value',
}

resulting in:

[
    "dbname='true'",
    "replication='true'",
]

?

[kibae]

"Replication" can have "database", "true" and "false" values.

https://github.com/postgres/postgres/blob/19dc233c32f2900e57b8da4f41c0f662ab42e080/src/backend/postmaster/postmaster.c#L2088

For a clean commit log, I will just clean up the connection related code and request a new pull.

[kibae]

Let me know if you have a problem with my code.

[brianc]

Hey @kibae - this is pretty awesome! Sorry for the delay. I'll merge this now!

What I'm thinking of doing is pulling out the Walstream code into it's own module & only adding the things to node-postgres itself to support the replication. That way the pg-walstream can live on its own. Would you be cool with me doing that or do you want to do that - I'm fine either way? I understand we'll still need the changes inside node-postgres core to support the replication setup & backend-events, but I try to keep things as slim as possible within node-postgres itself - having a separate installable module for the wal-stream will allow it to mature faster on its own.

Does that make sense?

[brianc]

I'll wait for @charmander's approval - looks like there might be some ⁉️ about the code still.

[kibae]

@brianc Oh, I see.
For a clean commit log, I will just clean up the connection related code and request a new pull.
And I'll separate project for logical replication :)