ჯავასკრიპტ - RXJS PostgreSQL ზეწოლა Node.js-ში

ორიგინალური სათაური: როგორ დავამუხრუჭოთ PgSQL შედეგების რიგების ნაკადი ჯავასკრიპტში?

მე განვიცდი არასაკმარისი მეხსიერების პრობლემას Node.js v4.5.0-ით RXJS(5.4.0) და PostgreSQL (დრაივერი "pg": "6.1.4") თან.

მე მენიურდ ვქმნი PgSQL სვეტების ობსერვაბელს ასე:

return Rx.Observable.create((subscriber) => {
    pool.connect().then((client: pg.Client) => {
        const stream:any = client.query(query.toParam());
        stream.on('row', (row) => {
            subscriber.next(row);
        });

        stream.on('end', () => {
            subscriber.complete();
            client.release();
        });
    });
});

შემდეგ მე ვუპირისპირებ რამდენიმე ოპერატორს rx ობსერვებლისთვის და ვასრულებ გარკვეულ დამუშავებას. შენიშვნა: მონაცემთა ბაზიდან დაბრუნებული რიგები ცოტა მძიმეა.

გამოძიება მეპატიჟება ერთ დასკვნამდე: მონაცემთა ბაზიდან ხაზები დაბრუნდება ბევრად სწრაფად, ვიდრე შეძლებს მათი დამუშავება. ამ მძიმე მონაცემებისთვის უნდა იყოს მეხსიერება რეზერვირებული, რათა დაელოდოს დამუშავებას, და ეს იწვევს მეხსიერების ამოწურვას:

საშინელი შეცდომა: CALL_AND_RETRY_LAST განაწილება ვერ მოხერხდა - პროცესს აღარ ყოფნის მეხსიერება

გაუქმების ხაფანგი: 6

მე ვერ ვხედავ PostgreSQL დრაივერში სტრიმის შესაჩერებელ ნებისმიერი ვარიანტს. გაქვთ იდეები, როგორ შემიძლია ვჭრიდე ეს პრობლემა?

პასუხი #1

თუ გამოიყენებთ ერთდროულად ავტორის მიერ შექმნილ pg-cursor ბიბლიოთეკას, ეს უნდა იყოს შედარებით მარტივი:

return Rx.Observable.defer(() => pg.connect())
  .flatMap(client => {
    const cursor = client.query(new Cursor('SELECT * FROM some_table WHERE prop > $1', [100]))

    const observableCursor = Rx.Observable.bindNodeCallback(cursor.read.bind(cursor));

    // Get the first 100 items
    observableCursor(100)
      .map(processRows)
      // This will only emit after the first one completes
      // and will recursively call this for each result
      .expand(_ => 
        observableCursor(100)
          .map(processRows)
      )
      // Unsubscribes once we don't get any more results
      .takeWhile(rows => rows.length > 0)
  });

პასუხი #2

შევეცადე რაც შეიძლება წაკითხადება გამეკეთებინა @paulpdaniels-ის მიერ შემოთავაზებული გადაწყვეტილება, მაგრამ დარწმუნებული არ ვარ, ეს ნამდვილად რეაქტიული სტილია თუ არა.

მე ასევე მაქვს სხვადასხვა დამუშავების ფუნქცია, რომელიც დამოკიდებულია კონტექსტურ ცვლადზე. მომწონს ისეთი არქიტექტურა, რომელიც ადვილად მატებს სხვადასხვა rxjs ოპერატორის ჯაჭვს, რომელიც დამოკიდებულია კონტექსტის პირობებზე და მალავს cursor.read ლოგიკას, მაგრამ მეშინია, რომ ეს შეუძლებელი იქნება.

უნდა არსებობდეს ადგილი, როდესაც მე ცხადად ვუთითებ მონაცემთა ბაზას შემდეგი მონაცემთა ნაკრების (cursor.read) წამოღებაზე და ეს უნდა იყოს ადგილი მას შემდეგ, რაც გავათავისუფლებ ჩემს მეხსიერებას მძიმე მონაცემებიდან.

შემდგომი გადაწყვეტილება არის TypeScript-ში:

const db = require('./libs/db.js');
import * as Cursor from 'pg-cursor';
import * as pg from 'pg';
import * as Rx from 'rxjs';

let cursor, readHandler;

const source = Rx.Observable.create((subscriber) => {
    db.getClient().then((client: pg.Client) => {

        cursor = client.query(new Cursor('SELECT * FROM some_table WHERE prop > $1', [100]));
        readHandler = function (err, rowsBulk) {
            if (err) {
                subscriber.error(err);
                return client.release();
            }
            if (!rowsBulk.length) {
                subscriber.complete();
                return client.release();
            }
            subscriber.next(rowsBulk);
        };
    }).then(() => {
        // fetch first 100 records
        cursor.read(100, readHandler);
    });
});

source.flatMap((item) => {
    return new Promise((resolve) => {
        console.log('processing batch of items');
        setTimeout(() => { // timeout simulates processing
            item = null;
            console.log('release memory');
            resolve({});
        }, 5000);
    });
}).map(() => {
    // fetch every next 100 records
    console.log('allocate new memory');
    cursor.read(100, readHandler);
    return '.';
}).subscribe(console.log, (err) => {
    console.log('Error occurred');
    console.error(err);
});