synchronization.service.ts 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115
  1. import { resolve } from "path";
  2. import { map, Observable, of, Subject } from "rxjs";
  3. import { BaseMessage } from "../dependencies/fisappmessagejsutilty/dependencies/dependencies";
  4. import { LoggingService } from "../dependencies/fisloggingservice/services/logging-service";
  5. import { MessageLog } from "../dependencies/fisloggingservice/type/datatype";
  6. import { MessageSynchronisationServiceInterface, MessageSynchronisationServiceSetting } from "../type/datatype";
  7. export class MessageSyncrhonizationService implements MessageSynchronisationServiceInterface {
  8. private settings: MessageSynchronisationServiceSetting
  9. private sourceSrc: LoggingService = new LoggingService()
  10. private targetSrc: LoggingService = new LoggingService()
  11. public init(settings: MessageSynchronisationServiceSetting): void {
  12. this.settings = settings;
  13. }
  14. // Incoming obstriger serves as a trigger point to perform another synchronization
  15. public subscribe(obsTrigger: Observable<string>): Observable<BaseMessage> {
  16. let msg: Subject<BaseMessage> = new Subject()
  17. obsTrigger.subscribe({
  18. next: obs => {
  19. let missingMsg = this.dataConversion()
  20. missingMsg.subscribe({
  21. next: element => {
  22. msg.next(element)
  23. }
  24. })
  25. }
  26. })
  27. // trigger by timer
  28. if (!obsTrigger) {
  29. this.dataConversion()
  30. }
  31. let result: Observable<BaseMessage> = msg.asObservable()
  32. return result
  33. }
  34. private dataConversion(): Observable<BaseMessage> {
  35. // let subjectOutput = this.syncrhonize()
  36. let obsOutput: Observable<BaseMessage> = this.syncrhonize().pipe(
  37. map((msg: MessageLog) => {
  38. // console.log(`Converting this ${msg.appData.msgId}`)
  39. return JSON.parse(<string>msg.appData.msgPayload)
  40. })
  41. )
  42. return obsOutput
  43. }
  44. private syncrhonize(): Subject<any> {
  45. let subjectOutput = new Subject()
  46. this.acquireData().then((data: { arr1: MessageLog[], arr2: MessageLog[] }) => {
  47. if (data.arr1.length === data.arr2.length) {
  48. console.log(`No synchronization needed`)
  49. } else {
  50. this.checkArrayDifferences(data).then((data: MessageLog[]) => {
  51. data.forEach(msgElement => {
  52. subjectOutput.next(msgElement)
  53. })
  54. })
  55. }
  56. }).catch((e) => console.error(e))
  57. return subjectOutput
  58. }
  59. // Acquires the available data from designated storage
  60. private async acquireData(): Promise<any> {
  61. const promiseQuery: Promise<any> = new Promise((resolve, reject) => {
  62. let allSets: {
  63. arr1: MessageLog[],
  64. arr2: MessageLog[]
  65. } = {
  66. arr1: [],
  67. arr2: []
  68. }
  69. let set1
  70. let set2
  71. this.sourceSrc.init(this.settings.incomingSource).then(()=>{
  72. this.targetSrc.init(this.settings.target).then(()=>{
  73. this.sourceSrc.filter({ msgTag: this.settings.incomingSource.tags[0] }).then((data: MessageLog[]) => {
  74. set1 = data
  75. }).then(() => {
  76. this.targetSrc.filter({ msgTag: this.settings.target.tags[0] }).then((data: MessageLog[]) => {
  77. set2 = data
  78. allSets.arr1 = set1
  79. allSets.arr2 = set2
  80. resolve(allSets)
  81. })
  82. })
  83. })
  84. })
  85. })
  86. return promiseQuery
  87. }
  88. // compare results and return differences`
  89. private async checkArrayDifferences(args: { arr1?: any[], arr2?: any[] }): Promise<any[]> {
  90. return new Promise((resolve, reject) => {
  91. let missingMsg: MessageLog[] = []
  92. args.arr1.forEach((msgElement: MessageLog) => {
  93. if (args.arr2.some(obj => obj.appData.msgId === msgElement.appData.msgId)) {
  94. console.log(`Item Found!`)
  95. } else {
  96. console.log(`This ${msgElement.appData.msgId} is not found`)
  97. missingMsg.push(msgElement)
  98. resolve(missingMsg)
  99. }
  100. })
  101. })
  102. }
  103. }