123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154 |
- /* ----------------------- TEST3A {Mongo to Mongo} ----------------------- */
- /* This test is focusing on comparing 2 different arrays of message logs from 2 different storage.
- Which is local file mongo as the control/source, and then comparing the data from cloud mongoDB
- server data, and then synchronizing them */
- import { Observable, map, Subject, takeUntil, take, of, timer, from } from "rxjs";
- import { BaseMessage } from "../dependencies/fisappmessagejsutilty/dependencies/dependencies";
- import { LogSetting, MessageLog } from "../dependencies/fisloggingservice/type/datatype";
- import { IncomingMessageService } from "../services/incomingMessage.service";
- import { MessageSyncrhonizationService } from "../services/synchronization.service";
- import { ErrorTrigger, MessageSynchronisationServiceSetting } from "../type/datatype";
- import { StreamingService } from "./test-streamOBS";
- /* Pre - Defined Data && Settings */
- // This service will stream the messages from the local testRequest.json messages
- // into the designated location that will be specified later.
- const stream = new StreamingService()
- /* Using the instance of the streaming declared earlier, we feed 4 messages into the
- subscribers that are going to subsscribe to this source_payload. Please note that
- source_payload will emite the messages stream from the instance of stream service
- and further feed them into the other Subject which is called source_payload_subject. */
- const source_synchronize = new MessageSyncrhonizationService()
- const source_incoming = new IncomingMessageService()
- const source_payload: Observable<BaseMessage> = stream.stream().pipe(take(4))
- const source_payload_subject: Subject<BaseMessage> = new Subject()
- source_payload.subscribe({
- next: (data) => {
- source_payload_subject.next(data)
- }
- })
- /* Same thing as the above. The only difference is the we feed only 2 messages
- to simulate streaming error. We want to see if it will sync the other 2 later
- on. But generall the declarative structure is the same as the above. */
- const target_incoming = new IncomingMessageService()
- const target_payload: Observable<BaseMessage> = stream.stream().pipe(take(2))
- const target_payload_subject: Subject<BaseMessage> = new Subject()
- target_payload.subscribe({
- next: (data) => {
- target_payload_subject.next(<BaseMessage>data)
- },
- error: e => console.error(e),
- complete: () => { `Target Payload Completed` }
- })
- /* Declare the designated database. I am using windev's mongo storage to store the data.
- Hence here, is the block that definte the target and it's associated specifications.
- This will be the target and will receive the predefined set of data to be logged as
- prepared earlier in the code above.s */
- let source_storage: LogSetting = {
- cacheMessageLimit: 0,
- storage: "MongoDB",
- setting: {
- appName: 'Default from client',
- appLocName: 'To be generated in client',
- logLocName: 'To be generated in client',
- },
- customSetting: {
- server: "192.168.100.59:27017",
- database: "test"
- }
- }
- // I forgot what this is for lol...
- let source_dataSet: LogSetting & { incomingObservable: Observable<BaseMessage> } = {
- storage: source_storage.storage,
- setting: source_storage.setting,
- customSetting: source_storage.customSetting,
- incomingObservable: source_payload_subject
- }
- /* Same as above. Also declaring another designated database. But this one will be used
- as the target for synching. For such I purposely push only half the of the completed
- dataset in order to test out the sync later. I am using my own cloud atlas mongo
- database on this. The address can always be changed. */
- let target_storage: LogSetting = {
- cacheMessageLimit: 0,
- storage: "MongoDB",
- setting: {
- appName: 'Default from client',
- appLocName: 'To be generated in client',
- logLocName: 'To be generated in client',
- },
- customSetting: {
- srv: true,
- user: "testDB",
- password: "h1nt1OyXw6QeUnzS",
- server: "cluster0.29sklte.mongodb.net",
- database: "log",
- }
- }
- let target_dataSet: LogSetting & { incomingObservable: Observable<BaseMessage> } = {
- storage: target_storage.storage,
- setting: target_storage.setting,
- customSetting: target_storage.customSetting,
- incomingObservable: target_payload_subject
- }
- // Combine source and target storage to form MessageSynchronisationServiceSetting
- let settings: MessageSynchronisationServiceSetting = {
- incomingSource: {
- //all of the settings to be combined here
- ...source_storage,
- tags: ['Incoming']
- }, //LogSetting & {tags:string[] },
- target: {
- ...target_storage,
- tags: ['Incoming']
- } //LogSetting & {tags:string[] }
- }
- /* -------- SYNCHRONIZATION --------- */
- // This is where the real test begin. THe process before this were merely just configuring
- // the settings of where to sync. Here the initial intialize data will first log the
- // messages into the designated database as specified earlier.
- function initializeData() { // To store the data into the designated databases.
- source_incoming.init(source_dataSet)
- target_incoming.init(target_dataSet)
- }
- // Done by appoximately 5-8 Seconds
- initializeData() // Call the function to store the data into the designated databases.
- source_synchronize.init(settings)
- /* This is where the synchronizatin happens. Upon initializin ghe target earlier, we
- will put in a trigger to execute the synching. It will check for missing datas and
- publish the missing messages accordingly. For this test purpose, we purposely set to
- 7 seconds to allow the logging process earlier to take place first before engaging
- in synching activity to prevent confusion and interference of the pre-logging stage. */
- setTimeout(() => {
- // This wil act as the trigger error. Although the definition of this error is
- // still subject for enhancements in the near future.
- let sampleError: ErrorTrigger = {
- status: 1,
- message: "NO. I dont want to work"
- }
- let triggerSync: Observable<ErrorTrigger> = from([sampleError])
- let sync = source_synchronize.subscribe(triggerSync)
- sync.subscribe({
- next: (msgToBeSynched) => {
- console.log(`synching ... ${msgToBeSynched.header.messageID}`)
- target_payload_subject.next(msgToBeSynched)
- }
- })
- }, 5000)
|