Skip to content

add maxChanges parameter #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jrs526 wants to merge 2 commits into
base: master
Choose a base branch
from
Open

add maxChanges parameter #14

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3e0617e
add maxChanges parameter
Mar 16, 2022
aadbcce
fix jsdoc
Mar 16, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 19 additions & 4 deletions index.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,15 @@
const EventEmitter = require('events');

const _getReadQuery = function(walOptions, slotName){
const _getReadQuery = function(walOptions, slotName, maxChanges){
let changesSql = '';
Object.keys(walOptions).forEach(function(option){
const value = walOptions[option];
changesSql += `, '${option}', '${value}'`;
});

const sql = `SELECT * FROM pg_catalog.pg_logical_slot_get_changes('${slotName}', NULL, NULL${changesSql});`;
if (maxChanges == undefined) maxChanges = "NULL";

const sql = `SELECT * FROM pg_catalog.pg_logical_slot_get_changes('${slotName}', NULL, ${maxChanges}${changesSql});`;

return {
text: sql,
Expand All @@ -27,8 +29,20 @@ const _init = async function(client, slotName, temporary){
}
};

/**
* Wal2JSONListener will emit 'changes' event containing the rows as returned
* from pg_catalog.pg_logical_slot_get_changes. This event may contain any number of rows.
*
* To limit the number of rows, use the maxChanges parameter. It is important to note
* that the maxChanges value is directly passed to pg_catalog.pg_logical_slot_get_changes as the
* upto_nchanges parameter, as documented at https://www.postgresql.org/docs/12/functions-admin.html.
* Specifically,
* > If upto_nchanges is non-NULL, decoding will stop when the number of rows produced by decoding
* > exceeds the specified value. Note, however, that the actual number of rows returned may be larger,
* > since this limit is only checked after adding the rows produced when decoding each new transaction commit.
*/
class Wal2JSONListener extends EventEmitter {
constructor(client, {slotName, timeout, temporary}, walOptions={}) {
constructor(client, {slotName, timeout, temporary, maxChanges}, walOptions={}) {
super();
this.slotName = slotName || 'test_slot';
this.walOptions = walOptions;
Expand All @@ -37,7 +51,8 @@ class Wal2JSONListener extends EventEmitter {
this.client = client;
this.timeout = timeout;
this.running = false;
this.readQuery = _getReadQuery(this.walOptions, this.slotName);
this.maxChanges = maxChanges
this.readQuery = _getReadQuery(this.walOptions, this.slotName, this.maxChanges);
this.client.connect();
}

Expand Down