Fis-AuditMessage/services/synchronization.service.ts

import { resolve } from "path";
import { Observable, of, Subject } from "rxjs";
import { BaseMessage } from "../dependencies/fisappmessagejsutilty/dependencies/dependencies";
import { LoggingService } from "../dependencies/fisloggingservice/services/logging-service";
import { MessageSynchronisationServiceInterface, MessageSynchronisationServiceSetting } from "../type/datatype";

export class MessageSyncrhonizationService implements MessageSynchronisationServiceInterface {

    settings: MessageSynchronisationServiceSetting
    sourceSrc: LoggingService = new LoggingService()
    targetSrc: LoggingService = new LoggingService()

    public async init(settings: MessageSynchronisationServiceSetting): Promise<void> {
        this.settings = settings;
        // Wrap promise so that init must be instantiated with resolve before proceeding with subscribe method
        let promiseInit: Promise<void> = new Promise((resolve, reject) => {
            try {
                this.sourceSrc.init(settings.incomingSource).then((data) => {
                    if (!data) reject()
                    // console.log(`File Storage:`)
                    // data.forEach(e => console.log(e.appLogLocId))
                }).then(() => {
                    this.targetSrc.init(settings.target).then((data) => {
                        if (!data) reject()
                        // console.log(`Mongo Storage`)
                        // data.forEach(e => console.log(e.appLogLocId))
                        resolve()
                    })
                })
            }
            catch (e) {
                console.error(e)
            }
        })
        return promiseInit
    }

    public async subscribe(obs: Observable<string>): Promise<Observable<any>> {
        let subjectOutput = new Subject()

        this.acquireData().then((data) => {
            this.compareResult(data).then((data) => {
                const obs : Observable<any> = of(...data)
                obs.subscribe(subjectOutput)
                // log remaining data?
                this.targetSrc.subscribe(obs)
            })
        }).catch((e) => console.error(e))

        return subjectOutput.asObservable()
    }

    // Acquires the available data from designated storage
    private async acquireData(): Promise<any> {
        const promiseQuery: Promise<any> = new Promise((resolve, reject) => {
            let allSets: any = {}
            let set1
            let set2
            this.sourceSrc.filter({ msgTag: this.settings.incomingSource.tags[0] }).then((data) => {
                set1 = data
            }).then(() => {
                this.targetSrc.filter({ msgTag: this.settings.target.tags[0] }).then((data) => {
                    set2 = data
                    allSets.arr1 = set1
                    allSets.arr2 = set2
                    resolve(allSets)
                })
            })
        })
        return promiseQuery
    }
    // compare results and return differences
    private async compareResult(args: any): Promise<any> {
        return new Promise((resolve, reject) => {
            let data = []
            args.arr1.forEach((element) => {
                if (args.arr2.some(obj => obj.appData.msgId === element.appData.msgId)) {
                    console.log(`Item Found!`)
                } else {
                    console.log(`This ${element.appData.msgId} is not found`)
                    data.push(element)
                    resolve(data)
                }
            })
        })
    }


}