synchronization.service.ts 4.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122
  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 async init(settings: MessageSynchronisationServiceSetting): Promise<void> {
  12. this.settings = settings;
  13. // Wrap promise so that init must be instantiated with resolve before proceeding with subscribe method
  14. let promiseInit: Promise<void> = new Promise((resolve, reject) => {
  15. this.sourceSrc.init(settings.incomingSource).then((data) => {
  16. if (!data) reject()
  17. }).then(() => {
  18. this.targetSrc.init(settings.target).then((data) => {
  19. if (!data) reject()
  20. resolve()
  21. })
  22. })
  23. })
  24. return promiseInit
  25. }
  26. // Incoming obstriger serves as a trigger point to perform another synchronization
  27. public subscribe(obsTrigger: Observable<string>): Observable<BaseMessage> {
  28. let msg : Subject<BaseMessage> = new Subject()
  29. obsTrigger.subscribe({
  30. next: obs => {
  31. let missingMsg = this.dataConversion()
  32. missingMsg.subscribe({
  33. next: element => {
  34. msg.next(element)
  35. }
  36. })
  37. }
  38. })
  39. // trigger by timer
  40. if(!obsTrigger) {
  41. this.dataConversion()
  42. }
  43. let result: Observable<BaseMessage> = msg.asObservable()
  44. return result
  45. }
  46. private dataConversion(): Observable<BaseMessage> {
  47. // let subjectOutput = this.syncrhonize()
  48. let obsOutput: Observable<BaseMessage> = this.syncrhonize().pipe(
  49. map((msg: MessageLog) => {
  50. // console.log(`Converting this ${msg.appData.msgId}`)
  51. return JSON.parse(<string>msg.appData.msgPayload)
  52. })
  53. )
  54. return obsOutput
  55. }
  56. private syncrhonize(): Subject<any> {
  57. let subjectOutput = new Subject()
  58. this.acquireData().then((data: { arr1: MessageLog[], arr2: MessageLog[] }) => {
  59. if (data.arr1.length === data.arr2.length) {
  60. console.log(`No synchronization needed`)
  61. } else {
  62. this.checkArrayDifferences(data).then((data: MessageLog[]) => {
  63. data.forEach(msgElement => {
  64. subjectOutput.next(msgElement)
  65. })
  66. })
  67. }
  68. }).catch((e) => console.error(e))
  69. return subjectOutput
  70. }
  71. // Acquires the available data from designated storage
  72. private async acquireData(): Promise<any> {
  73. const promiseQuery: Promise<any> = new Promise((resolve, reject) => {
  74. let allSets: {
  75. arr1: MessageLog[],
  76. arr2: MessageLog[]
  77. } = {
  78. arr1: [],
  79. arr2: []
  80. }
  81. let set1
  82. let set2
  83. this.sourceSrc.filter({ msgTag: this.settings.incomingSource.tags[0] }).then((data: MessageLog[]) => {
  84. set1 = data
  85. }).then(() => {
  86. this.targetSrc.filter({ msgTag: this.settings.target.tags[0] }).then((data: MessageLog[]) => {
  87. set2 = data
  88. allSets.arr1 = set1
  89. allSets.arr2 = set2
  90. resolve(allSets)
  91. })
  92. })
  93. })
  94. return promiseQuery
  95. }
  96. // compare results and return differences`
  97. private async checkArrayDifferences(args: { arr1?: any[], arr2?: any[] }): Promise<any[]> {
  98. return new Promise((resolve, reject) => {
  99. let missingMsg: MessageLog[] = []
  100. args.arr1.forEach((msgElement: MessageLog) => {
  101. if (args.arr2.some(obj => obj.appData.msgId === msgElement.appData.msgId)) {
  102. console.log(`Item Found!`)
  103. } else {
  104. console.log(`This ${msgElement.appData.msgId} is not found`)
  105. missingMsg.push(msgElement)
  106. resolve(missingMsg)
  107. }
  108. })
  109. })
  110. }
  111. }