feat(core): add support for read connections (#116)

Now users can specify multiple read connections via `replicas` option. You can provide only fields that differ from master connection, rest will be taken from it.

By default select queries will use random read connection if not inside transaction. You can specify connection type manually in `EntityManager.getConnection(type: 'read' | 'write')`.

Closes #77

lib/EntityManager.ts

@@ -1,7 +1,7 @@
 import { Configuration, RequestContext, Utils, ValidationError } from './utils';
 import { EntityAssigner, EntityFactory, EntityLoader, EntityRepository, EntityValidator, ReferenceType } from './entity';
 import { LockMode, UnitOfWork } from './unit-of-work';
-import { FilterQuery, IDatabaseDriver } from './drivers';
+import { AbstractSqlDriver, FilterQuery, IDatabaseDriver } from './drivers';
 import { EntityData, EntityMetadata, EntityName, IEntity, IEntityType, IPrimaryKey } from './decorators';
 import { QueryBuilder, QueryOrderMap, SmartQueryHelper } from './query';
 import { MetadataStorage } from './metadata';
@@ -24,8 +24,8 @@ export class EntityManager {
     return this.driver as D;
   }
 
-  getConnection<C extends Connection = Connection>(): C {
-    return this.driver.getConnection() as C;
+  getConnection<C extends Connection = Connection>(type?: 'read' | 'write'): C {
+    return this.driver.getConnection(type) as C;
   }
 
   getRepository<T extends IEntityType<T>>(entityName: EntityName<T>): EntityRepository<T> {
@@ -44,9 +44,9 @@ export class EntityManager {
     return this.validator;
   }
 
-  createQueryBuilder(entityName: EntityName<IEntity>, alias?: string): QueryBuilder {
+  createQueryBuilder(entityName: EntityName<IEntity>, alias?: string, type?: 'read' | 'write'): QueryBuilder {
     entityName = Utils.className(entityName);
-    return new QueryBuilder(entityName, this.metadata, this.driver, this.transactionContext, alias);
+    return new QueryBuilder(entityName, this.metadata, this.driver as AbstractSqlDriver, this.transactionContext, alias, type);
   }
 
   async find<T extends IEntityType<T>>(entityName: EntityName<T>, where?: FilterQuery<T>, options?: FindOptions): Promise<T[]>;

lib/MikroORM.ts

@@ -45,8 +45,7 @@ export class MikroORM {
   }
 
   async connect(): Promise<IDatabaseDriver> {
-    const connection = this.driver.getConnection();
-    await connection.connect();
+    const connection = await this.driver.connect();
     const clientUrl = connection.getClientUrl();
     const dbName = this.config.get('dbName');
     this.logger.log('info', `MikroORM successfully connected to database ${chalk.green(dbName)}${clientUrl ? ' on ' + chalk.green(clientUrl) : ''}`);
@@ -59,7 +58,7 @@ export class MikroORM {
   }
 
   async close(force = false): Promise<void> {
-    return this.driver.getConnection().close(force);
+    return this.driver.close(force);
   }
 
   getMetadata(): MetadataStorage {

lib/connections/Connection.ts

@@ -3,16 +3,22 @@ import { Transaction as KnexTransaction } from 'knex';
 import chalk from 'chalk';
 import highlight from 'cli-highlight';
 
-import { Configuration, Utils } from '../utils';
+import { Configuration, ConnectionOptions, Utils } from '../utils';
 import { MetadataStorage } from '../metadata';
 
 export abstract class Connection {
 
-  protected readonly logger = this.config.getLogger();
   protected metadata: MetadataStorage;
   protected abstract client: any;
 
-  constructor(protected readonly config: Configuration) { }
+  constructor(protected readonly config: Configuration,
+              protected readonly options?: ConnectionOptions,
+              protected readonly type: 'read' | 'write' = 'write') {
+    if (!this.options) {
+      const props = ['dbName', 'clientUrl', 'host', 'port', 'user', 'password', 'multipleStatements', 'pool'] as const;
+      this.options = props.reduce((o, i) => { (o[i] as any) = this.config.get(i); return o; }, {} as ConnectionOptions);
+    }
+  }
 
   /**
    * Establishes connection to database
@@ -42,12 +48,12 @@ export abstract class Connection {
 
   getConnectionOptions(): ConnectionConfig {
     const ret: ConnectionConfig = {};
-    const url = new URL(this.config.getClientUrl());
-    ret.host = this.config.get('host', url.hostname);
-    ret.port = this.config.get('port', +url.port);
-    ret.user = this.config.get('user', url.username);
-    ret.password = this.config.get('password', url.password);
-    ret.database = this.config.get('dbName', url.pathname.replace(/^\//, ''));
+    const url = new URL(this.options!.clientUrl || this.config.getClientUrl());
+    this.options!.host = ret.host = this.config.get('host', url.hostname);
+    this.options!.port = ret.port = this.config.get('port', +url.port);
+    this.options!.user = ret.user = this.config.get('user', url.username);
+    this.options!.password = ret.password = this.config.get('password', url.password);
+    this.options!.dbName = ret.database = this.config.get('dbName', url.pathname.replace(/^\//, ''));
 
     return ret;
   }
@@ -76,7 +82,12 @@ export abstract class Connection {
       query = highlight(query, { language, ignoreIllegals: true, theme: this.config.getHighlightTheme() });
     }
 
-    const msg = `${query}` + (Utils.isDefined(took) ? chalk.grey(` [took ${took} ms]`) : '');
+    let msg = query + (Utils.isDefined(took) ? chalk.grey(` [took ${chalk.grey(took)} ms]`) : '');
+
+    if (this.config.get('replicas', []).length > 0) {
+      msg += chalk.cyan(` (via ${this.type} connection '${this.options!.name || this.config.get('name') || this.options!.host}')`);
+    }
+
     this.config.getLogger().log('query', msg);
   }
 

lib/drivers/AbstractSqlDriver.ts

@@ -93,7 +93,7 @@ export abstract class AbstractSqlDriver<C extends AbstractSqlConnection = Abstra
   }
 
   protected createQueryBuilder(entityName: string, ctx?: Transaction): QueryBuilder {
-    return new QueryBuilder(entityName, this.metadata, this, ctx);
+    return new QueryBuilder(entityName, this.metadata, this, ctx, undefined, ctx ? 'write' : 'read');
   }
 
   protected extractManyToMany<T extends IEntityType<T>>(entityName: string, data: EntityData<T>): EntityData<T> {

lib/drivers/DatabaseDriver.ts

@@ -2,14 +2,15 @@ import { FilterQuery, IDatabaseDriver } from './IDatabaseDriver';
 import { EntityData, EntityMetadata, EntityProperty, IEntity, IEntityType, IPrimaryKey } from '../decorators';
 import { MetadataStorage } from '../metadata';
 import { Connection, QueryResult, Transaction } from '../connections';
-import { Configuration, Utils } from '../utils';
+import { Configuration, ConnectionOptions, Utils } from '../utils';
 import { QueryOrder, QueryOrderMap } from '../query';
 import { Platform } from '../platforms';
 import { LockMode } from '../unit-of-work';
 
 export abstract class DatabaseDriver<C extends Connection> implements IDatabaseDriver<C> {
 
   protected readonly connection: C;
+  protected readonly replicas: C[] = [];
   protected readonly platform: Platform;
   protected readonly logger = this.config.getLogger();
   protected metadata: MetadataStorage;
@@ -74,8 +75,26 @@ export abstract class DatabaseDriver<C extends Connection> implements IDatabaseD
     return ret as T;
   }
 
-  getConnection(): C {
-    return this.connection as C;
+  async connect(): Promise<C> {
+    await this.connection.connect();
+    await Promise.all(this.replicas.map(replica => replica.connect()));
+
+    return this.connection;
+  }
+
+  getConnection(type: 'read' | 'write' = 'write'): C {
+    if (type === 'write' || this.replicas.length === 0) {
+      return this.connection as C;
+    }
+
+    const rand = Utils.randomInt(0, this.replicas.length - 1);
+
+    return this.replicas[rand] as C;
+  }
+
+  async close(force?: boolean): Promise<void> {
+    await Promise.all(this.replicas.map(replica => replica.close(force)));
+    await this.connection.close(force);
   }
 
   getPlatform(): Platform {
@@ -92,4 +111,17 @@ export abstract class DatabaseDriver<C extends Connection> implements IDatabaseD
     return meta ? meta.primaryKey : this.config.getNamingStrategy().referenceColumnName();
   }
 
+  protected createReplicas(cb: (c: ConnectionOptions) => C): C[] {
+    const replicas = this.config.get('replicas', [])!;
+    const ret: C[] = [];
+    const props = ['dbName', 'clientUrl', 'host', 'port', 'user', 'password', 'multipleStatements', 'pool', 'name'] as const;
+
+    replicas.forEach((conf: Partial<ConnectionOptions>) => {
+      props.forEach(prop => (conf[prop] as any) = prop in conf ? conf[prop] : this.config.get(prop));
+      ret.push(cb(conf as ConnectionOptions));
+    });
+
+    return ret;
+  }
+
 }

lib/drivers/IDatabaseDriver.ts

@@ -7,7 +7,11 @@ import { MetadataStorage } from '../metadata';
 
 export interface IDatabaseDriver<C extends Connection = Connection> {
 
-  getConnection(): C;
+  connect(): Promise<C>;
+
+  close(force?: boolean): Promise<void>;
+
+  getConnection(type?: 'read' | 'write'): C;
 
   /**
    * Finds selection of entities

lib/drivers/MongoDriver.ts

@@ -15,7 +15,7 @@ export class MongoDriver extends DatabaseDriver<MongoConnection> {
 
   async find<T extends IEntityType<T>>(entityName: string, where: FilterQuery<T>, populate: string[], orderBy: QueryOrderMap, limit: number, offset: number): Promise<T[]> {
     where = this.renameFields(entityName, where);
-    const res = await this.connection.find<T>(entityName, where, orderBy, limit, offset);
+    const res = await this.getConnection('read').find<T>(entityName, where, orderBy, limit, offset);
 
     return res.map((r: T) => this.mapResult<T>(r, this.metadata.get(entityName))!);
   }
@@ -26,19 +26,19 @@ export class MongoDriver extends DatabaseDriver<MongoConnection> {
     }
 
     where = this.renameFields(entityName, where) as FilterQuery<T>;
-    const res = await this.connection.find<T>(entityName, where, orderBy, 1, undefined, fields);
+    const res = await this.getConnection('read').find<T>(entityName, where, orderBy, 1, undefined, fields);
 
     return this.mapResult<T>(res[0], this.metadata.get(entityName));
   }
 
   async count<T extends IEntityType<T>>(entityName: string, where: FilterQuery<T>): Promise<number> {
     where = this.renameFields(entityName, where);
-    return this.connection.countDocuments<T>(entityName, where);
+    return this.getConnection('read').countDocuments<T>(entityName, where);
   }
 
   async nativeInsert<T extends IEntityType<T>>(entityName: string, data: EntityData<T>): Promise<QueryResult> {
     data = this.renameFields(entityName, data);
-    return this.connection.insertOne<EntityData<T>>(entityName, data);
+    return this.getConnection('write').insertOne<EntityData<T>>(entityName, data);
   }
 
   async nativeUpdate<T extends IEntityType<T>>(entityName: string, where: FilterQuery<T> | IPrimaryKey, data: EntityData<T>): Promise<QueryResult> {
@@ -49,7 +49,7 @@ export class MongoDriver extends DatabaseDriver<MongoConnection> {
     where = this.renameFields(entityName, where) as FilterQuery<T>;
     data = this.renameFields(entityName, data);
 
-    return this.connection.updateMany<T>(entityName, where, data);
+    return this.getConnection('write').updateMany<T>(entityName, where, data);
   }
 
   async nativeDelete<T extends IEntityType<T>>(entityName: string, where: FilterQuery<T> | IPrimaryKey): Promise<QueryResult> {
@@ -59,11 +59,11 @@ export class MongoDriver extends DatabaseDriver<MongoConnection> {
 
     where = this.renameFields(entityName, where) as FilterQuery<T>;
 
-    return this.connection.deleteMany<T>(entityName, where);
+    return this.getConnection('write').deleteMany<T>(entityName, where);
   }
 
   async aggregate(entityName: string, pipeline: any[]): Promise<any[]> {
-    return this.connection.aggregate(entityName, pipeline);
+    return this.getConnection('read').aggregate(entityName, pipeline);
   }
 
   private renameFields(entityName: string, data: any): any {

lib/drivers/MySqlDriver.ts

@@ -5,6 +5,7 @@ import { MySqlPlatform } from '../platforms/MySqlPlatform';
 export class MySqlDriver extends AbstractSqlDriver<MySqlConnection> {
 
   protected readonly connection = new MySqlConnection(this.config);
+  protected readonly replicas = this.createReplicas(conf => new MySqlConnection(this.config, conf, 'read'));
   protected readonly platform = new MySqlPlatform();
 
 }

lib/query/QueryBuilder.ts

@@ -6,8 +6,7 @@ import { EntityProperty, IEntity } from '../decorators';
 import { ReferenceType } from '../entity';
 import { QueryFlag, QueryOrderMap, QueryType } from './enums';
 import { LockMode } from '../unit-of-work';
-import { AbstractSqlConnection } from '../connections/AbstractSqlConnection';
-import { IDatabaseDriver } from '../drivers';
+import { AbstractSqlDriver } from '../drivers';
 import { MetadataStorage } from '../metadata';
 
 /**
@@ -33,16 +32,16 @@ export class QueryBuilder {
   private _limit: number;
   private _offset: number;
   private lockMode?: LockMode;
-  private readonly connection = this.driver.getConnection() as AbstractSqlConnection;
   private readonly platform = this.driver.getPlatform();
-  private readonly knex = this.connection.getKnex();
+  private readonly knex = this.driver.getConnection(this.connectionType).getKnex();
   private readonly helper = new QueryBuilderHelper(this.entityName, this.alias, this._aliasMap, this.metadata, this.knex, this.platform);
 
   constructor(private readonly entityName: string,
               private readonly metadata: MetadataStorage,
-              private readonly driver: IDatabaseDriver,
+              private readonly driver: AbstractSqlDriver,
               private readonly context?: Transaction,
-              readonly alias = `e0`) { }
+              readonly alias = `e0`,
+              private readonly connectionType?: 'read' | 'write') { }
 
   select(fields: string | string[], distinct = false): this {
     this._fields = Utils.asArray(fields);
@@ -236,7 +235,8 @@ export class QueryBuilder {
   }
 
   async execute(method: 'all' | 'get' | 'run' = 'all', mapResults = true): Promise<any> {
-    const res = await this.connection.execute(this.getKnexQuery(), [], method);
+    const type = this.connectionType || (method === 'run' ? 'write' : 'read');
+    const res = await this.driver.getConnection(type).execute(this.getKnexQuery(), [], method);
     const meta = this.metadata.get(this.entityName);
 
     if (!mapResults) {
@@ -251,7 +251,7 @@ export class QueryBuilder {
   }
 
   clone(): QueryBuilder {
-    const qb = new QueryBuilder(this.entityName, this.metadata, this.driver, this.context, this.alias);
+    const qb = new QueryBuilder(this.entityName, this.metadata, this.driver, this.context, this.alias, this.connectionType);
     Object.assign(qb, this);
 
     // clone array/object properties
@@ -410,6 +410,7 @@ export class QueryBuilder {
     if (this.context) {
       qb.transacting(this.context);
     }
+
     return qb;
   }
 

lib/unit-of-work/UnitOfWork.ts

@@ -102,7 +102,7 @@ export class UnitOfWork {
     const promise = async (tx: Transaction) => await Utils.runSerial(this.changeSets, changeSet => this.commitChangeSet(changeSet, tx));
 
     if (runInTransaction) {
-      await this.em.getConnection().transactional(trx => promise(trx));
+      await this.em.getConnection('write').transactional(trx => promise(trx));
     } else {
       await promise(this.em.getTransactionContext());
     }

lib/utils/Configuration.ts

@@ -31,6 +31,7 @@ export class Configuration {
     hydrator: ObjectHydrator,
     tsNode: false,
     debug: false,
+    verbose: false,
     cache: {
       enabled: true,
       adapter: FileCacheAdapter,
@@ -75,8 +76,8 @@ export class Configuration {
     this.init();
   }
 
-  get<T extends keyof MikroORMOptions, U extends MikroORMOptions[T]>(key: T, defaultValue?: U): MikroORMOptions[T] {
-    return (Utils.isDefined(this.options[key]) ? this.options[key] : defaultValue) as MikroORMOptions[T];
+  get<T extends keyof MikroORMOptions, U extends MikroORMOptions[T]>(key: T, defaultValue?: U): U {
+    return (Utils.isDefined(this.options[key]) ? this.options[key] : defaultValue) as U;
   }
 
   set<T extends keyof MikroORMOptions, U extends MikroORMOptions[T]>(key: T, value: U): void {
@@ -172,8 +173,19 @@ export class Configuration {
 
 }
 
-export interface MikroORMOptions {
+export interface ConnectionOptions {
+  name?: string;
   dbName: string;
+  clientUrl?: string;
+  host?: string;
+  port?: number;
+  user?: string;
+  password?: string;
+  multipleStatements?: boolean; // for mysql driver
+  pool: PoolConfig;
+}
+
+export interface MikroORMOptions extends ConnectionOptions {
   entities: (EntityClass<IEntity> | EntityClassGroup<IEntity>)[];
   entitiesDirs: string[];
   entitiesDirsTs: string[];
@@ -185,13 +197,7 @@ export interface MikroORMOptions {
   namingStrategy?: { new (): NamingStrategy };
   hydrator: { new (factory: EntityFactory, driver: IDatabaseDriver): Hydrator };
   entityRepository: { new (em: EntityManager, entityName: EntityName<IEntity>): EntityRepository<IEntity> };
-  clientUrl?: string;
-  host?: string;
-  port?: number;
-  user?: string;
-  password?: string;
-  multipleStatements?: boolean; // for mysql driver
-  pool: PoolConfig;
+  replicas?: Partial<ConnectionOptions>[];
   strict: boolean;
   logger: (message: string) => void;
   debug: boolean | LoggerNamespace[];

lib/utils/Utils.ts

@@ -269,4 +269,8 @@ export class Utils {
     }, [] as T[]);
   }
 
+  static randomInt(min: number, max: number): number {
+    return Math.round(Math.random() * (max - min)) + min;
+  }
+
 }