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 103 104 105 106 107 108 109 110 111 112 113 114 | 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 11x 10x 10x 10x 10x 10x 9x 9x 8x 7x 8x 14x 7x 9x 9x 18x 18x 18x 18x 9x 9x 4x 3x 14x 14x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x | 'use strict';
import {Client} from 'pg';
import {MikroORM} from '@mikro-orm/postgresql';
import {BackendRegistry} from '@owservable/core';
import PostgresBackend from './postgres.backend';
import PostgresListener from './postgres.listener';
import PostgresTablesEntitiesMap from './tables.entities.map';
import installPostgresTriggers from './functions/install.triggers';
import {PostgresLiveUpdatesRegistry} from './decorators/live.updates';
const SCHEMA_SYNC_LOCK_KEY: number = 776277001;
export type PostgresConnectorOptionsType = {
entities: any[];
host: string;
port: number;
user: string;
password: string;
dbName: string;
ssl?: boolean | Record<string, any>;
channel?: string;
updateSchema?: boolean;
safe?: boolean;
triggers?: boolean;
ormOptions?: any;
};
export default class PostgresConnector {
public static async init(options: PostgresConnectorOptionsType): Promise<any> {
if (PostgresConnector._orm) return PostgresConnector._orm;
const {entities, host, port, user, password, dbName, ssl, channel = 'owservable', updateSchema = true, safe = true, triggers = true, ormOptions = {}} = options;
const orm: any = await MikroORM.init({
entities,
host,
port,
user,
password,
dbName,
...(ssl ? {driverOptions: {ssl}} : {}),
...ormOptions
});
console.log('[@owservable/postgres] -> PostgreSQL connected to', `${host}:${port}/${dbName}`);
const connectionConfig: any = {host, port, user, password, database: dbName, ...(ssl ? {ssl} : {})};
if (updateSchema || triggers) {
await PostgresConnector._withSchemaLock(connectionConfig, async (): Promise<void> => {
if (updateSchema) {
await orm.schema.update({safe});
console.log('[@owservable/postgres] -> PostgreSQL schema synchronized', safe ? '(safe mode)' : '');
}
if (triggers) {
const liveEntities: any[] = entities.filter((entity: any): boolean => PostgresLiveUpdatesRegistry.has(entity));
await installPostgresTriggers(orm, liveEntities, channel);
}
});
}
const listener: PostgresListener = PostgresListener.init(connectionConfig, channel);
for (const entity of entities) {
const meta: any = orm.getMetadata().get(entity.name);
const tableName: string = meta.tableName;
PostgresTablesEntitiesMap.addTableToEntityMapping(tableName, entity);
BackendRegistry.register(tableName, new PostgresBackend(orm, entity, listener));
}
PostgresConnector._orm = orm;
return orm;
}
public static get orm(): any {
return PostgresConnector._orm;
}
public static em(): any {
return PostgresConnector._orm?.em.fork();
}
public static async close(): Promise<void> {
await PostgresListener.stop();
if (PostgresConnector._orm) {
await PostgresConnector._orm.close();
PostgresConnector._orm = null;
}
}
private static _orm: any;
private static async _withSchemaLock(config: any, execute: () => Promise<void>): Promise<void> {
const lockClient: Client = new Client(config);
await lockClient.connect();
try {
await lockClient.query('SELECT pg_advisory_lock($1)', [SCHEMA_SYNC_LOCK_KEY]);
await execute();
} finally {
try {
await lockClient.query('SELECT pg_advisory_unlock($1)', [SCHEMA_SYNC_LOCK_KEY]);
} finally {
await lockClient.end();
}
}
}
private constructor() {}
}
|