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
 
 
11x
11x
 
 
 
 
 
 
 
4x
4x
 
 
 
1x
 
1x
 
1x
1x
 
 
 
 
 
 
11x
11x
 
'use strict';
 
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 {
		if (!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 => {
			const {ns, documentKey, operationType, updateDescription, fullDocument} = change;
			this.next({ns, documentKey, operationType, updateDescription, fullDocument});
		});
	}
}
 
const observableDatabase = (): Subject<any> => ObservableDatabase.init();
export default observableDatabase;