123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139 |
- import { map, Observable, of, Subject } from "rxjs";
- import { BaseMessage } from "../dependencies/fisappmessagejsutilty/dependencies/dependencies";
- import { LoggingService } from "../dependencies/fisloggingservice/services/logging-service";
- import { LogSetting, MessageLog } from "../dependencies/fisloggingservice/type/datatype";
- import { ErrorTrigger, MessageAuditorServiceInterface, MessageSynchronisationServiceSetting } from "../type/datatype";
- export class MessageAuditorService implements MessageAuditorServiceInterface {
- private settings: MessageSynchronisationServiceSetting
- private sourceSrc: LoggingService = new LoggingService()
- private targetSrc: LoggingService = new LoggingService()
-
- public init(settings: MessageSynchronisationServiceSetting): void {
- this.settings = settings;
- }
-
- public subscribe(obsTrigger: Observable<ErrorTrigger>): Observable<BaseMessage> {
-
- let msg: Subject<BaseMessage> = new Subject()
-
- obsTrigger.subscribe({
- next: obsTrigger => {
- console.log(obsTrigger.message)
- let missingMsg = this.dataConversion()
- missingMsg.subscribe({
- next: element => {
- msg.next(element)
- }
- })
- }
- })
-
-
-
-
- let result: Observable<BaseMessage> = msg.asObservable()
- return result
- }
-
- private dataConversion(): Observable<BaseMessage> {
-
- let obsOutput: Observable<BaseMessage> = this.synchronize().pipe(
- map((msg: MessageLog) => {
- console.log(`Converting this ${msg.appData.msgId}`)
- return JSON.parse(<string>msg.appData.msgPayload)
- })
- )
- return obsOutput
- }
-
- private synchronize(): Subject<any> {
- let subjectOutput = new Subject()
-
- this.acquireData().then((data: { arr1: MessageLog[], arr2: MessageLog[] }) => {
-
-
- if (data.arr1.length === data.arr2.length) {
- console.log(`No synchronization needed`)
- } else {
-
-
- this.checkArrayDifferences(data).then((data: MessageLog[]) => {
- data.forEach(msgElement => {
-
-
- subjectOutput.next(msgElement)
- })
- })
- }
- }).catch((e) => console.error(e))
- return subjectOutput
- }
-
- private async acquireData(): Promise<any> {
- const promiseQuery: Promise<any> = new Promise((resolve, reject) => {
-
- let allSets: {
- arr1: MessageLog[],
- arr2: MessageLog[]
- } = {
- arr1: [],
- arr2: []
- }
- let set1
- let set2
-
- this.sourceSrc.init(this.settings.incomingSource).then(() => {
- this.targetSrc.init(this.settings.target).then(() => {
-
- this.sourceSrc.filter({ msgTag: this.settings.incomingSource.tags[0] }).then((data: MessageLog[]) => {
- set1 = data
- }).then(() => {
- this.targetSrc.filter({ msgTag: this.settings.target.tags[0] }).then((data: MessageLog[]) => {
- set2 = data
- allSets.arr1 = set1
- allSets.arr2 = set2
- resolve(allSets)
- })
- })
- })
- })
- })
- return promiseQuery
- }
-
- private async checkArrayDifferences(args: { arr1?: any[], arr2?: any[] }): Promise<any[]> {
- return new Promise((resolve, reject) => {
- let missingMsg: MessageLog[] = []
- args.arr1.forEach((msgElement: MessageLog) => {
-
-
-
- if (args.arr2.some(obj => obj.appData.msgId === msgElement.appData.msgId)) {
- console.log(`Item Found!`)
- } else {
- console.log(`This ${msgElement.appData.msgId} is not found`)
- missingMsg.push(msgElement)
- resolve(missingMsg)
- }
- })
- })
- }
- }
|