Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | 3x 3x 3x 3x 14x 14x 6x 20x 13x 13x 13x 13x 13x 13x 13x 13x 13x 48x 47x 1x 1x 13x 13x 48x 47x 47x 47x 47x 46x 4x 4x 46x 46x 1x 1x 47x 13x 13x 13x | 'use strict';
import {ReplaySubject, Subject} from 'rxjs';
export type SqliteNotificationType = {
table: string;
op: string;
id: string;
changed: string[] | null;
};
const JOURNAL_TABLE: string = '_owservable_changes';
const PRUNE_EVERY_POLLS: number = 40;
export default class SqliteJournalPoller {
public static init(orm: any, pollIntervalMs: number = 250): SqliteJournalPoller {
if (!SqliteJournalPoller._instance) SqliteJournalPoller._instance = new SqliteJournalPoller(orm, pollIntervalMs);
return SqliteJournalPoller._instance;
}
public static get instance(): SqliteJournalPoller {
return SqliteJournalPoller._instance;
}
public static stop(): void {
if (!SqliteJournalPoller._instance) return;
SqliteJournalPoller._instance._stop();
SqliteJournalPoller._instance = null;
}
public readonly notifications: Subject<SqliteNotificationType>;
public readonly lifecycle: ReplaySubject<any>;
private readonly _orm: any;
private _interval: any;
private _lastSeen: number;
private _pollCount: number;
private _polling: boolean;
private static _instance: SqliteJournalPoller;
private constructor(orm: any, pollIntervalMs: number) {
this._orm = orm;
this._lastSeen = 0;
this._pollCount = 0;
this._polling = false;
this.notifications = new Subject<SqliteNotificationType>();
this.lifecycle = new ReplaySubject<any>(1);
this._interval = setInterval((): void => {
this._poll() //
.then((): null => null)
.catch((error: any): void => {
console.error('[@owservable/sqlite] -> SqliteJournalPoller poll error:', error);
this.lifecycle.next({type: 'error', timestamp: new Date(), error});
});
}, pollIntervalMs);
this.lifecycle.next({type: 'live', timestamp: new Date()});
console.log(`[@owservable/sqlite] -> SqliteJournalPoller polling "${JOURNAL_TABLE}" every ${pollIntervalMs}ms`);
}
private async _poll(): Promise<void> {
if (this._polling) return;
this._polling = true;
try {
const connection: any = this._orm.em.getConnection();
const rows: any[] = await connection.execute(`SELECT id, table_name, op, pk, changed FROM "${JOURNAL_TABLE}" WHERE id > ? ORDER BY id ASC`, [this._lastSeen]);
for (const row of rows) {
this._lastSeen = row.id;
this.notifications.next({
table: String(row.table_name),
op: String(row.op),
id: String(row.pk),
changed: row.changed ? String(row.changed).split(',').filter(Boolean) : null
});
}
this._pollCount++;
if (this._pollCount >= PRUNE_EVERY_POLLS) {
this._pollCount = 0;
await connection.execute(`DELETE FROM "${JOURNAL_TABLE}" WHERE id <= ?`, [this._lastSeen]);
}
} finally {
this._polling = false;
}
}
private _stop(): void {
clearInterval(this._interval);
this._interval = null;
this.lifecycle.next({type: 'close', timestamp: new Date()});
}
}
|