All files / src/mongodb/functions observable.database.ts

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
 
 
2x
2x
2x
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
2x
2x
 
'use strict';
 
import {pick} from 'lodash';
import {Subject} from 'rxjs';
import mongoose from 'mongoose';
import {ChangeStream} from 'mongodb';
 
class ObservableDatabase extends Subject<any> {
	private readonly _stream: ChangeStream;
	private static _instance: ObservableDatabase;
 
	public static init(): ObservableDatabase {
		Iif (!ObservableDatabase._instance) ObservableDatabase._instance = new ObservableDatabase();
		return ObservableDatabase._instance;
	}
 
	constructor() {
		super();
 
		const db: mongoose.mongo.Db = mongoose.connection.db;
 
		this._stream = db.watch([], {fullDocument: 'updateLookup'});
		this._stream.on('change', (change: any): void => {
			this.next(pick(change, ['ns', 'documentKey', 'operationType', 'updateDescription', 'fullDocument']));
		});
	}
}
 
const observableDatabase = (): Subject<any> => ObservableDatabase.init();
export default observableDatabase;