Consuming a stream with pg-cursor

main.mjs

import { Readable } from 'stream'
import pg from 'pg'
import Cursor from 'pg-cursor'

class TailStream extends Readable {
    constructor(client, text, values) {
        super({
            highWaterMark: 5,
            objectMode: true
        });
        this.client = client;
        this.cursor = client.query(new Cursor(text, values));
        this.pendingReads = 0;

        this.cursor.on('row', (row) => {
            this.pendingReads -= 1;
            this.push(row);
        });
    }
    _read(n) {
        if (this.pendingReads == 0) {
            this.pendingReads = n;
            console.log('reading the next batch of', n, 'records');
            this.cursor.read(n, (err) => {
                if (err) {
                    this.destroy(err);
                }
            });
        }
    }
    _destroy(err, callback) {
        this.cursor.close(() => {
            this.client.release()
            callback()
        })
    }
}

async function main() {
    const client = new pg.Client('postgres://materialize@localhost:6875/materialize');
    await client.connect();

    const stream = new TailStream(client, "TAIL foo", []);

    for await (const row of stream) {
        console.log('tail row', row);
    }
}

main();