synchronization.service.ts 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106
  1. import { resolve } from "path";
  2. import { 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. settings: MessageSynchronisationServiceSetting
  9. sourceSrc: LoggingService = new LoggingService()
  10. 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. try {
  16. this.sourceSrc.init(settings.incomingSource).then((data) => {
  17. if (!data) reject()
  18. }).then(() => {
  19. this.targetSrc.init(settings.target).then((data) => {
  20. if (!data) reject()
  21. resolve()
  22. })
  23. })
  24. }
  25. catch (e) {
  26. console.error(e)
  27. }
  28. })
  29. return promiseInit
  30. }
  31. // Incoming obstriger serves as a trigger point to perform another synchronization
  32. public subscribe(obsTrigger: Observable<string>): Observable<any> {
  33. let subjectOutput = this.syncrhonize()
  34. // obs.subscribe({
  35. // next: element => {
  36. // subjectOutput = this.syncrhonize()
  37. // }
  38. // })
  39. return subjectOutput.asObservable()
  40. }
  41. private syncrhonize(): Subject<any> {
  42. let subjectOutput = new Subject()
  43. this.acquireData().then((data: { arr1: MessageLog[], arr2: MessageLog[] }) => {
  44. if (data.arr1.length === data.arr2.length) {
  45. console.log(`No synchronization needed`)
  46. } else {
  47. this.checkArrayDifferences(data).then((data: MessageLog[]) => {
  48. data.forEach(msgElement => {
  49. subjectOutput.next(msgElement)
  50. })
  51. })
  52. }
  53. }).catch((e) => console.error(e))
  54. return subjectOutput
  55. }
  56. // Acquires the available data from designated storage
  57. private async acquireData(): Promise<any> {
  58. const promiseQuery: Promise<any> = new Promise((resolve, reject) => {
  59. let allSets: {
  60. arr1: MessageLog[],
  61. arr2: MessageLog[]
  62. } = {
  63. arr1: [],
  64. arr2: []
  65. }
  66. let set1
  67. let set2
  68. this.sourceSrc.filter({ msgTag: this.settings.incomingSource.tags[0] }).then((data: MessageLog[]) => {
  69. set1 = data
  70. }).then(() => {
  71. this.targetSrc.filter({ msgTag: this.settings.target.tags[0] }).then((data: MessageLog[]) => {
  72. set2 = data
  73. allSets.arr1 = set1
  74. allSets.arr2 = set2
  75. resolve(allSets)
  76. })
  77. })
  78. })
  79. return promiseQuery
  80. }
  81. // compare results and return differences
  82. private async checkArrayDifferences(args: { arr1?: any[], arr2?: any[] }): Promise<any[]> {
  83. return new Promise((resolve, reject) => {
  84. let missingMsg: MessageLog[] = []
  85. args.arr1.forEach((msgElement: MessageLog) => {
  86. if (args.arr2.some(obj => obj.appData.msgId === msgElement.appData.msgId)) {
  87. console.log(`Item Found!`)
  88. } else {
  89. console.log(`This ${msgElement.appData.msgId} is not found`)
  90. missingMsg.push(msgElement)
  91. resolve(missingMsg)
  92. }
  93. })
  94. })
  95. }
  96. }