|
@@ -1,15 +1,15 @@
|
|
|
-/* ----------------------- TEST3A {Mongo to Mongo} ----------------------- */
|
|
|
+/* ----------------------- TEST3C {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 { BaseMessage, ResponseMessage } 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";
|
|
|
+import { MessageAuditorService } from "../services/message-auditor.service";
|
|
|
+import { LoggingService } from "../dependencies/fisloggingservice/services/logging-service";
|
|
|
|
|
|
/* Pre - Defined Data && Settings */
|
|
|
// This service will stream the messages from the local testRequest.json messages
|
|
@@ -20,25 +20,25 @@ const stream = new StreamingService()
|
|
|
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({
|
|
|
+const publisher_sync = new MessageAuditorService()
|
|
|
+const publisher_Log = new LoggingService()
|
|
|
+const publisher_take_four_messages: Observable<BaseMessage> = stream.stream().pipe(take(4))
|
|
|
+const publisher: Subject<BaseMessage> = new Subject()
|
|
|
+publisher_take_four_messages.subscribe({
|
|
|
next: (data) => {
|
|
|
- source_payload_subject.next(data)
|
|
|
+ publisher.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({
|
|
|
+const subscriber_log = new LoggingService()
|
|
|
+const subscriber_take_two_messagse: Observable<BaseMessage> = stream.stream().pipe(take(2))
|
|
|
+const subscriber: Subject<BaseMessage> = new Subject()
|
|
|
+subscriber_take_two_messagse.subscribe({
|
|
|
next: (data) => {
|
|
|
- target_payload_subject.next(<BaseMessage>data)
|
|
|
+ subscriber.next(<ResponseMessage>data)
|
|
|
},
|
|
|
error: e => console.error(e),
|
|
|
complete: () => { `Target Payload Completed` }
|
|
@@ -49,7 +49,7 @@ target_payload.subscribe({
|
|
|
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 = {
|
|
|
+let publisher_storage: LogSetting = {
|
|
|
cacheMessageLimit: 0,
|
|
|
storage: "MongoDB",
|
|
|
setting: {
|
|
@@ -63,19 +63,11 @@ let source_storage: LogSetting = {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-// 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 = {
|
|
|
+let subscriber_storage: LogSetting = {
|
|
|
cacheMessageLimit: 0,
|
|
|
storage: "MongoDB",
|
|
|
setting: {
|
|
@@ -92,22 +84,15 @@ let target_storage: LogSetting = {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-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,
|
|
|
+ ...publisher_storage,
|
|
|
tags: ['Incoming']
|
|
|
}, //LogSetting & {tags:string[] },
|
|
|
target: {
|
|
|
- ...target_storage,
|
|
|
+ ...subscriber_storage,
|
|
|
tags: ['Incoming']
|
|
|
} //LogSetting & {tags:string[] }
|
|
|
}
|
|
@@ -117,25 +102,28 @@ let settings: MessageSynchronisationServiceSetting = {
|
|
|
// 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)
|
|
|
+ publisher_Log.init(publisher_storage).then(() => {
|
|
|
+ publisher_Log.subscribe(publisher)
|
|
|
+ })
|
|
|
+ subscriber_log.init(subscriber_storage).then(() => {
|
|
|
+ subscriber_log.subscribe(subscriber)
|
|
|
+ })
|
|
|
}
|
|
|
|
|
|
// Done by appoximately 5-8 Seconds
|
|
|
initializeData() // Call the function to store the data into the designated databases.
|
|
|
-source_synchronize.init(settings)
|
|
|
+publisher_sync.init(settings)
|
|
|
|
|
|
/* This is where the synchronization logic is called. The errorSubject will act as a trigger
|
|
|
mechanism to execute the synchronization. */
|
|
|
-
|
|
|
let errorSubject: Subject<ErrorTrigger> = new Subject()
|
|
|
// Subscribe to errorSubject notification
|
|
|
-let sync = source_synchronize.subscribe(errorSubject)
|
|
|
+let sync = publisher_sync.subscribe(errorSubject)
|
|
|
sync.subscribe({
|
|
|
next: (msgToBeSynched) => {
|
|
|
console.log(`synching ... ${msgToBeSynched.header.messageID}`)
|
|
|
// the missing data returned will be pushed (next(message)) into the target payload.
|
|
|
- target_payload_subject.next(msgToBeSynched)
|
|
|
+ subscriber.next(msgToBeSynched)
|
|
|
}
|
|
|
})
|
|
|
|