test3c.ts 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175
  1. /* ----------------------- TEST3C {Mongo to Mongo} ----------------------- */
  2. /* This test is focusing on comparing 2 different arrays of message logs from 2 different storage.
  3. Which is local file mongo as the control/source, and then comparing the data from cloud mongoDB
  4. server data, and then synchronizing them */
  5. import { Observable, map, Subject, takeUntil, take, of, timer, from } from "rxjs";
  6. import { BaseMessage, ResponseMessage } from "../dependencies/fisappmessagejsutilty/dependencies/dependencies";
  7. import { LogSetting, MessageLog } from "../dependencies/fisloggingservice/type/datatype";
  8. import { ErrorTrigger, MessageSynchronisationServiceSetting } from "../type/datatype";
  9. import { StreamingService } from "./test-streamOBS";
  10. import { MessageAuditorService } from "../services/message-auditor.service";
  11. import { LoggingService } from "../dependencies/fisloggingservice/services/logging-service";
  12. /* Pre - Defined Data && Settings */
  13. // This service will stream the messages from the local testRequest.json messages
  14. // into the designated location that will be specified later.
  15. const stream = new StreamingService()
  16. /* Using the instance of the streaming declared earlier, we feed 4 messages into the
  17. subscribers that are going to subsscribe to this source_payload. Please note that
  18. source_payload will emite the messages stream from the instance of stream service
  19. and further feed them into the other Subject which is called source_payload_subject. */
  20. const publisher_sync = new MessageAuditorService()
  21. const publisher_Log = new LoggingService()
  22. const publisher_take_four_messages: Observable<BaseMessage> = stream.stream().pipe(take(4))
  23. const publisher: Subject<BaseMessage> = new Subject()
  24. publisher_take_four_messages.subscribe({
  25. next: (data) => {
  26. publisher.next(data)
  27. }
  28. })
  29. /* Same thing as the above. The only difference is the we feed only 2 messages
  30. to simulate streaming error. We want to see if it will sync the other 2 later
  31. on. But generall the declarative structure is the same as the above. */
  32. const subscriber_log = new LoggingService()
  33. const subscriber_take_two_messagse: Observable<BaseMessage> = stream.stream().pipe(take(2))
  34. const subscriber: Subject<BaseMessage> = new Subject()
  35. subscriber_take_two_messagse.subscribe({
  36. next: (data) => {
  37. subscriber.next(<ResponseMessage>data)
  38. },
  39. error: e => console.error(e),
  40. complete: () => { `Target Payload Completed` }
  41. })
  42. /* Declare the designated database. I am using windev's mongo storage to store the data.
  43. Hence here, is the block that definte the target and it's associated specifications.
  44. This will be the target and will receive the predefined set of data to be logged as
  45. prepared earlier in the code above.s */
  46. let publisher_storage: LogSetting = {
  47. cacheMessageLimit: 0,
  48. storage: "MongoDB",
  49. setting: {
  50. appName: 'Default from client',
  51. appLocName: 'To be generated in client',
  52. logLocName: 'To be generated in client',
  53. },
  54. customSetting: {
  55. server: "192.168.100.59:27017",
  56. database: "test"
  57. }
  58. }
  59. /* Same as above. Also declaring another designated database. But this one will be used
  60. as the target for synching. For such I purposely push only half the of the completed
  61. dataset in order to test out the sync later. I am using my own cloud atlas mongo
  62. database on this. The address can always be changed. */
  63. let subscriber_storage: LogSetting = {
  64. cacheMessageLimit: 0,
  65. storage: "MongoDB",
  66. setting: {
  67. appName: 'Default from client',
  68. appLocName: 'To be generated in client',
  69. logLocName: 'To be generated in client',
  70. },
  71. customSetting: {
  72. srv: true,
  73. user: "testDB",
  74. password: "h1nt1OyXw6QeUnzS",
  75. server: "cluster0.29sklte.mongodb.net",
  76. database: "log",
  77. }
  78. }
  79. // Combine source and target storage to form MessageSynchronisationServiceSetting
  80. let settings: MessageSynchronisationServiceSetting = {
  81. incomingSource: {
  82. //all of the settings to be combined here
  83. ...publisher_storage,
  84. tags: ['Incoming']
  85. }, //LogSetting & {tags:string[] },
  86. target: {
  87. ...subscriber_storage,
  88. tags: ['Incoming']
  89. } //LogSetting & {tags:string[] }
  90. }
  91. /* -------- SYNCHRONIZATION --------- */
  92. // This is where the real test begin. THe process before this were merely just configuring
  93. // the settings of where to sync. Here the initial intialize data will first log the
  94. // messages into the designated database as specified earlier.
  95. function initializeData() { // To store the data into the designated databases.
  96. publisher_Log.init(publisher_storage).then(() => {
  97. publisher_Log.subscribe(publisher)
  98. })
  99. subscriber_log.init(subscriber_storage).then(() => {
  100. subscriber_log.subscribe(subscriber)
  101. })
  102. }
  103. // Done by appoximately 5-8 Seconds
  104. initializeData() // Call the function to store the data into the designated databases.
  105. publisher_sync.init(settings)
  106. /* This is where the synchronization logic is called. The errorSubject will act as a trigger
  107. mechanism to execute the synchronization. */
  108. let errorSubject: Subject<ErrorTrigger> = new Subject()
  109. // Subscribe to errorSubject notification
  110. let sync = publisher_sync.subscribe(errorSubject)
  111. sync.subscribe({
  112. next: (msgToBeSynched) => {
  113. console.log(`synching ... ${msgToBeSynched.header.messageID}`)
  114. // the missing data returned will be pushed (next(message)) into the target payload.
  115. subscriber.next(msgToBeSynched)
  116. }
  117. })
  118. // Set time oout for 5 seconds to allow the initial logging stage to complete it's logging
  119. // implementation first before proceedint to trigger the sync
  120. setTimeout(() => {
  121. // This wil act as the trigger error.Although the definition of this error is
  122. // still subject for enhancements in the near future.
  123. let sampleError: ErrorTrigger = {
  124. status: 1,
  125. message: "NO. I dont want to work"
  126. }
  127. errorSubject.next(sampleError)
  128. }, 5000)
  129. /* THis is testing for generating error message to be fed into the error subject
  130. to act as additional trigger to exectute the synchronization when there's no internet
  131. connection. */
  132. const dns = require('dns');
  133. // Function to check internet connectivity. Basically just look up the site of example.com
  134. // using the built in libray of DNS.
  135. function checkInternetConnectivity() {
  136. dns.lookup('example.com', (err) => {
  137. if (err && err.code === 'ENOTFOUND') {
  138. let errorMsg: ErrorTrigger = {
  139. status: 0,
  140. message: `No internet connection`
  141. }
  142. errorSubject.next(errorMsg)
  143. } else {
  144. // Emit a message indicating internet connectivity
  145. // console.log('Internet connection is available');
  146. }
  147. });
  148. }
  149. // Interval time (in milliseconds) for checking connectivity
  150. const intervalTime = 1000; // Check every 1 second
  151. // Start checking connectivity at intervals
  152. const interval = setInterval(checkInternetConnectivity, intervalTime);
  153. // Stop checking connectivity after a certain duration (e.g., 1 minute)
  154. const duration = 60000; // 1 minute
  155. setTimeout(function () {
  156. clearInterval(interval);
  157. console.log('Internet connectivity monitoring stopped');
  158. }, duration);