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 99 100 101 102 | 5x 5x 5x 19x 19x 19x 19x 19x 19x 19x 19x 19x 19x 45x 32x 19x 9x 7x 6x 1x 19x 4x 1x 3x 1x 10x 7x 7x 7x 7x 1x 6x 6x 5x 1x 4x 4x 4x 2x 2x 2x 2x 2x 4x 5x | 'use strict';
import {each, get} from 'lodash';
import {Subject} from 'rxjs';
import {filter, skip} from 'rxjs/operators';
import PostgresListener, {PostgresNotificationType} from '../postgres.listener';
class PostgresObservableTable extends Subject<any> {
private readonly _orm: any;
private readonly _entity: any;
private readonly _tableName: string;
private readonly _pkProperty: string;
private readonly _pkIsNumeric: boolean;
private readonly _columnsToProperties: Map<string, string>;
constructor(orm: any, entity: any, listener: PostgresListener) {
super();
this._orm = orm;
this._entity = entity;
const meta: any = orm.getMetadata().get(entity.name);
this._tableName = meta.tableName;
this._pkProperty = meta.primaryKeys[0];
const pkProp: any = meta.properties[this._pkProperty];
this._pkIsNumeric = ['number', 'integer', 'bigint', 'smallint', 'float', 'double', 'decimal'].includes(String(pkProp?.type ?? '').toLowerCase());
this._columnsToProperties = new Map<string, string>();
each(meta.props, (prop: any): void => {
each(prop.fieldNames ?? [], (fieldName: string): void => {
this._columnsToProperties.set(fieldName, prop.name);
});
});
listener.notifications //
.pipe(filter((notification: PostgresNotificationType): boolean => notification?.table === this._tableName))
.subscribe({
next: (notification: PostgresNotificationType): void => {
this._process(notification) //
.then((): null => null)
.catch((error: any): void => {
console.error(`[@owservable/postgres] -> PostgresObservableTable[${this._tableName}] Error processing notification:`, {notification, error});
});
}
});
listener.lifecycle //
.pipe(filter((event: any): boolean => 'live' === event?.type))
.pipe(skip(1))
.subscribe({
next: (): void => this.next({})
});
}
public get tableName(): string {
return this._tableName;
}
public get pkProperty(): string {
return this._pkProperty;
}
public coercePrimaryKey(id: string): any {
return this._pkIsNumeric ? Number(id) : id;
}
private async _process(notification: PostgresNotificationType): Promise<void> {
const {op: operationType, id, changed} = notification;
const ns: any = {coll: this._tableName};
const documentKey: any = {_id: id};
if ('delete' === operationType) {
return this.next({ns, documentKey, operationType});
}
const em: any = this._orm.em.fork();
const entity: any = await em.findOne(this._entity, {[this._pkProperty]: this.coercePrimaryKey(id)});
if (!entity) {
return this.next({ns, documentKey, operationType});
}
const {wrap} = require('@mikro-orm/core');
const fullDocument: any = wrap(entity).toObject();
let updateDescription: any;
if ('update' === operationType) {
const updatedFields: any = {};
each(changed ?? [], (column: string): void => {
const property: string = this._columnsToProperties.get(column) ?? column;
updatedFields[property] = get(fullDocument, property, true);
});
updateDescription = {updatedFields, removedFields: []};
}
this.next({ns, documentKey, operationType, updateDescription, fullDocument});
}
}
export default PostgresObservableTable;
|