some cleaning

documents/monkey.json.txt

@@ -0,0 +1 @@
+{"theme":"sonokai","themeLight":"serika","themeDark":"serika_dark","autoSwitchTheme":false,"customTheme":false,"customThemeColors":["#323437","#e2b714","#e2b714","#646669","#2c2e31","#d1d0c5","#ca4754","#7e2a33","#ca4754","#7e2a33"],"favThemes":[],"showKeyTips":true,"showLiveWpm":true,"showTimerProgress":true,"smoothCaret":"medium","quickRestart":"tab","punctuation":false,"numbers":false,"words":50,"time":30,"mode":"quote","quoteLength":[3],"language":"english","fontSize":1.5,"freedomMode":true,"difficulty":"normal","blindMode":false,"quickEnd":false,"caretStyle":"default","paceCaretStyle":"default","flipTestColors":false,"layout":"default","funbox":"none","confidenceMode":"off","indicateTypos":"off","timerStyle":"mini","colorfulMode":false,"randomTheme":"off","timerColor":"main","timerOpacity":"1","stopOnError":"off","showAllLines":false,"keymapMode":"react","keymapStyle":"staggered","keymapLegendStyle":"dynamic","keymapLayout":"overrideSync","keymapShowTopRow":"layout","fontFamily":"Roboto_Mono","smoothLineScroll":false,"alwaysShowDecimalPlaces":false,"alwaysShowWordsHistory":false,"singleListCommandLine":"manual","capsLockWarning":true,"playSoundOnError":"off","playSoundOnClick":"5","soundVolume":"0.5","startGraphsAtZero":true,"showOutOfFocusWarning":true,"paceCaret":"off","paceCaretCustomSpeed":100,"repeatedPace":true,"pageWidth":"max","accountChart":["on","on","on","on"],"minWpm":"off","minWpmCustomSpeed":100,"highlightMode":"letter","typingSpeedUnit":"wpm","ads":"result","hideExtraLetters":false,"strictSpace":true,"minAcc":"custom","minAccCustom":90,"showLiveAcc":true,"showLiveBurst":true,"monkey":false,"repeatQuotes":"off","oppositeShiftMode":"off","customBackground":"","customBackgroundSize":"cover","customBackgroundFilter":[0,1,1,1,1],"customLayoutfluid":"qwerty#dvorak#colemak","monkeyPowerLevel":"off","minBurst":"flex","minBurstCustomSpeed":100,"burstHeatmap":false,"britishEnglish":false,"lazyMode":false,"showAverage":"both","tapeMode":"off"}

index.ts

@@ -1,6 +1,5 @@
 import { Subject, from } from 'rxjs';
 import * as fs from 'fs'
-export * from './services/fis.retransmission.service';
 export * from './services/server-client.service';
 
 const messagesJSON: any = fs.readFileSync('payload.json')

interfaces/general.interface.ts

@@ -2,12 +2,6 @@
 
 import { Observable, Subject } from "rxjs"
 
-
-export enum ColorCode {
-    'GREEN' = 'GREEN',
-    'YELLOW' = 'YELLOW',
-    'RED' = 'RED'
-}
 export interface ConnectionState {
     status: 'BUFFER' | 'DIRECT_PUBLISH';
     reason?: string;
@@ -29,11 +23,7 @@ export interface ServerResponse {
     msgId: string
 }
 
-export interface ReportStatus {
-    code: ColorCode,
-    message: string,
-    payload?: any,
-}
+
 // https://grpc.io/docs/what-is-grpc/core-concepts/
 export interface GrpcConnectionType {
     instanceType: '' | 'server' | 'client'
@@ -44,7 +34,7 @@ export interface Message {
     message: MessageLog | string
 }
 
-export type Status = -1 | 0 | 1 // For status chain effect
+export type State = -1 | 0 | 1 // For status chain effect
 
 
 export interface ConnectionAttribute {
@@ -86,3 +76,9 @@ export interface ConnectionID {
     local: string,
     remote: string
 }
+
+export interface OutGoingInfo {
+    StreamID: string,
+    PublisherID: string,
+    SubscriberID: string
+}

services/buffer.service.ts

@@ -1,176 +1,214 @@
 // bufferService.ts
-import { BehaviorSubject, Observable, Subject, from, map, switchMap } from 'rxjs';
-import mongoose, { Connection, Model, Document } from 'mongoose';
-import { ConnectionState, Message, MessageLog } from '../interfaces/general.interface';
-import { resolve } from 'path';
+import {
+  BehaviorSubject,
+  Observable,
+  Subject,
+  from,
+  map,
+  switchMap,
+} from "rxjs";
+import mongoose, { Connection, Model, Document } from "mongoose";
+import {
+  ConnectionState,
+  Message,
+  MessageLog,
+} from "../interfaces/general.interface";
 
 export class BufferService {
-    private messageStream: Subject<Message>
-    private connectionState: BehaviorSubject<ConnectionState>
-    private messageBuffer: Message[] = [];
-    private messageModel: Model<Message> | undefined;
-    private readonly dbUrl: string = process.env.MONGO as string;
-
-
-    constructor(
-        messageFromApp: Subject<Message>,
-        connectionStateSubject: BehaviorSubject<ConnectionState>,
-        dbName: string
-    ) {
-        this.messageStream = messageFromApp;
-        this.connectionState = connectionStateSubject;
-        this.setupSubscriptions(); // Note: The handle buffer will push the data in local array before pushing to mongo via initial check up model
-
-        /* Disable for now. Use local array first */
-        this.initializeDatabaseConnection(dbName).then((connection: mongoose.Connection) => {
-            const grpcMessageSchema = require('../models/message.schema');
-            this.messageModel = connection.model<Message>('Message', grpcMessageSchema)
-            this.transferLocalBufferToMongoDB() // transfer all data from local array into mongodb after the mongo setup is complete
-        }).catch(error => {
-            console.error('Database initialization failed:', error);
-            // Implement retry logic or additional error handling here
-        });
+  private messageStream: Subject<Message>;
+  private connectionState: BehaviorSubject<ConnectionState>;
+  private messageBuffer: Message[] = [];
+  private messageModel: Model<Message> | undefined;
+  private readonly dbUrl: string = process.env.MONGO as string;
+
+  constructor(
+    messageFromApp: Subject<Message>,
+    connectionStateSubject: BehaviorSubject<ConnectionState>,
+    dbName: string
+  ) {
+    this.messageStream = messageFromApp;
+    this.connectionState = connectionStateSubject;
+    this.setupSubscriptions(); // Note: The handle buffer will push the data in local array before pushing to mongo via initial check up model
+
+    // this.initializeDatabaseConnection(dbName)
+    //   .then((connection: mongoose.Connection) => {
+    //     const grpcMessageSchema = require("../models/message.schema");
+    //     this.messageModel = connection.model<Message>(
+    //       "Message",
+    //       grpcMessageSchema
+    //     );
+    //     this.transferLocalBufferToMongoDB(); // transfer all data from local array into mongodb after the mongo setup is complete
+    //   })
+    //   .catch((error) => {
+    //     console.error("Database initialization failed:", error);
+    //     // Implement retry logic or additional error handling here. Perhaps retry logic in the future...
+    //   });
+  }
+
+  // to be exposed to acquire the messages
+  public getMessages(): Observable<Message> {
+    return this.messageStream as Observable<Message>;
+  }
+
+  private setupSubscriptions(): void {
+    this.messageStream.subscribe({
+      next: (message: Message) => this.handleIncomingMessage(message),
+      error: (err) =>
+        console.error("Error in messageToBePublished subject:", err),
+      complete: () =>
+        console.log("messageToBePublished subscription completed"),
+    });
+
+    this.connectionState.subscribe({
+      next: (state: ConnectionState) => this.handleConnectionStateChanges(state),
+      error: (err) => console.error("Error in connectionState subject:", err),
+      complete: () => console.log("connectionState subscription completed"),
+    });
+  }
+
+  private async initializeDatabaseConnection(
+    dbName: string
+  ): Promise<Connection> {
+    try {
+      console.log(`${this.dbUrl}${dbName}`);
+      const connection: mongoose.Connection = await mongoose.createConnection(
+        `${this.dbUrl}${dbName}`
+      );
+      console.log(`Connected to ${this.dbUrl}${dbName}`);
+      return connection;
+    } catch (error) {
+      console.error("Error connecting to MongoDB:", error);
+      throw error;
     }
+  }
 
-    public getMessages(): Observable<Message> {
-        return this.messageStream as Observable<Message>
+  private handleIncomingMessage(message: Message): void {
+    if (this.connectionState.getValue().status === "BUFFER") {
+      this.bufferMessage(message);
     }
-
-    private setupSubscriptions(): void {
-        this.messageStream.subscribe({
-            next: (message: Message) => this.handleIncomingMessage(message),
-            error: (err) => console.error('Error in messageToBePublished subject:', err),
-            complete: () => console.log('messageToBePublished subscription completed')
-        });
-
-        this.connectionState.subscribe({
-            next: (state: ConnectionState) => this.handleConnectionStateChanges(state),
-            error: (err) => console.error('Error in connectionState subject:', err),
-            complete: () => console.log('connectionState subscription completed')
+    if (this.connectionState.getValue().status === "DIRECT_PUBLISH") {
+      /* Note: Since the main outGoingMessage is being published irregardless
+      of the connection state, so there's no need to do anything aside from
+      releasing buffered messages which will be handled by handleConnectionStateChange */
+      // additional logic here
+    }
+  }
+
+  private handleConnectionStateChanges(state: ConnectionState): void {
+    console.log(this.connectionState.getValue().status);
+    if (state.status === "BUFFER") {
+      if (state.payload && typeof state.payload !== "string") {
+        this.bufferMessage(state.payload); // Buffer the last message immediately
+      }
+    }
+    if (state.status === "DIRECT_PUBLISH") {
+      // Relese the messages by inserting them into the outgoing Messages together.
+      this.releaseBufferedMessages(this.messageStream);
+    }
+  }
+
+  private async bufferMessage(message: Message): Promise<void> {
+    if (this.messageModel) {
+      try {
+        // const newMessage = new this.messageModel(message);
+        await this.messageModel.create(message);
+        this.messageModel.countDocuments({}).then((count) => {
+          console.log(`Message${(message.message as MessageLog).appData.msgId} saved to MongoDB buffer. There is ${count} messages in datatbase at the moment.`);
         });
+      } catch (error) {
+        console.error("Error saving message to MongoDB:", error);
+        // Implement retry logic or additional error handling here 
+      }
+    } else {
+      this.messageBuffer.push(message); // Fallback to local buffer if model is not defined
+      console.log(`pushing ${(message.message as MessageLog).appData.msgId} into local array buffer.... There is now ${this.messageBuffer.length} messages`);
     }
+  }
+
+  private releaseBufferedMessages(
+    messageFromBuffer: Subject<Message>
+  ): Promise<boolean> {
+    return new Promise((resolve, reject) => {
+      if (this.messageModel) {
+        this.messageModel.countDocuments({}).then((count) => {
+          console.log(`There is ${count} messages in database buffer at the moment. Releasing them....`);
+        });
+        const stream = this.messageModel.find().cursor();
 
-    private async initializeDatabaseConnection(dbName: string): Promise<Connection> {
-        try {
-            console.log(`${this.dbUrl}${dbName}`)
-            const connection: mongoose.Connection = await mongoose.createConnection(`${this.dbUrl}${dbName}`);
-            console.log(`Connected to ${this.dbUrl}${dbName}`)
-            return connection;
-        } catch (error) {
-            console.error('Error connecting to MongoDB:', error);
-            throw error;
-        }
-    }
+        stream.on("data", async (message) => {
+          // Process each message individually`
+          messageFromBuffer.next(message);
+        });
 
-    private handleIncomingMessage(message: Message): void {
-        if (this.connectionState.getValue().status === 'BUFFER') {
-            this.bufferMessage(message);
-        }
-        if (this.connectionState.getValue().status === 'DIRECT_PUBLISH') {
-            // additional logic here
-        }
-    }
+        stream.on("error", (error) => {
+          console.error("Error streaming messages from MongoDB:", error);
+          reject(error);
+        });
 
-    private handleConnectionStateChanges(state: ConnectionState): void {
-        console.log(this.connectionState.getValue().status)
-        if (state.status === 'BUFFER') {
-            if (state.payload && typeof state.payload !== 'string') {
-                this.bufferMessage(state.payload); // Buffer the last message immediately
+        stream.on("end", async () => {
+          // Delete the data once it has been streamed
+          try {
+            if (this.messageModel) {
+              await this.messageModel.deleteMany({});
+              console.log("Data in Mongo deleted successfully.");
+            } else {
+              console.log(`Message Mongoose Model is not intiated properly...`);
             }
+          } catch (err) {
+            console.error("Error deleting data:", err);
+          }
+          resolve(true);
+        });
+      }
+      if (!this.messageModel) {
+        // If MongoDB model is not defined, use the local buffer
+        console.log(`Releasing buffer Message: currently there is ${this.messageBuffer.length} messages to be released`);
+        this.messageBuffer.forEach((message) =>
+          this.messageStream.next(message)
+        );
+        this.messageBuffer.length = 0; // Clear the local buffer after transferring
+        if (this.messageBuffer.length < 1) {
+          resolve(true);
+        } else {
+          reject(`Somehow the array is not emptied. This should not happen`);
         }
-        if (state.status === 'DIRECT_PUBLISH') {
-            this.releaseBufferedMessages(this.messageStream)
-        }
-    }
-
-    private async bufferMessage(message: Message): Promise<void> {
-        if (this.messageModel) {
+      }
+    });
+  }
+
+  public getStateObservable(): BehaviorSubject<ConnectionState> {
+    return this.connectionState;
+  }
+
+  private async transferLocalBufferToMongoDB(): Promise<void> {
+    return new Promise((resolve, reject) => {
+      console.log(`Transferring local array buffered Message: currently there is ${this.messageBuffer.length}. Transferring to database...`);
+      if (this.messageModel) {
+        let locallyBufferedMessage: Observable<Message> = from(this.messageBuffer);
+        locallyBufferedMessage.subscribe({
+          next: async (message: Message) => {
             try {
-                // const newMessage = new this.messageModel(message);
+              if (this.messageModel) {
                 await this.messageModel.create(message);
-                this.messageModel.countDocuments({}).then((count) => {
-                    console.log(`Message${(message.message as MessageLog).appData.msgId} saved to MongoDB buffer. There is ${count} messages in datatbase at the moment.`);
-                })
+                console.log(`Transferring ${(message.message as MessageLog).appData.msgId} into database.`);
+              }
             } catch (error) {
-                console.error('Error saving message to MongoDB:', error);
-                // Implement retry logic or additional error handling here
+              console.error("Error transferring message to MongoDB:", error);
             }
-        } else {
-            this.messageBuffer.push(message); // Fallback to local buffer if model is not defined
-            console.log(`pushing ${(message.message as MessageLog).appData.msgId} into local array buffer.... There is now ${this.messageBuffer.length} messages`)
-        }
-    }
-
-
-    private releaseBufferedMessages(messageFromBuffer: Subject<Message>): Promise<boolean> {
-        return new Promise((resolve, reject) => {
+          },
+          error: (err) => console.error(err),
+          complete: () => {
             if (this.messageModel) {
-                this.messageModel.countDocuments({}).then((count) => {
-                    console.log(`There is ${count} messages in datatbase buffer at the moment. Releasing them....`);
-
-                })
-                const stream = this.messageModel.find().cursor();
-
-                stream.on('data', async (message) => {
-                    // Process each message individually
-                    messageFromBuffer.next(message);
-                });
-
-                stream.on('error', (error) => {
-                    console.error('Error streaming messages from MongoDB:', error);
-                    reject(error)
-                });
-
-                stream.on('end', async () => {
-                    // Delete the data once it has been streamed
-                    try {
-                        if (this.messageModel) {
-                            await this.messageModel.deleteMany({});
-                            console.log('Data in Mongo deleted successfully.');
-                        } else {
-                            console.log(`Message Mongoose Model is not intiated properly...`)
-                        }
-                    } catch (err) {
-                        console.error('Error deleting data:', err);
-                    }
-                    resolve(true)
-                });
-            }
-            if (!this.messageModel) {
-                // If MongoDB model is not defined, use the local buffer
-                console.log(`Releasing buffer Message: currently there is ${this.messageBuffer.length} messages to be released`)
-                this.messageBuffer.forEach(message => this.messageStream.next(message));
-                this.messageBuffer.length = 0 // Clear the local buffer after transferring
-                if (this.messageBuffer.length < 1) {
-                    resolve(true)
-                } else {
-                    reject(`Somehow the array is not emptied. This should not happen`)
-                }
+              this.messageModel.countDocuments({}).then((count) => {
+                console.log(`Local buffered message transfer completed. There is a total of ${count} messages in database at the moment.`)
+                this.messageBuffer = [] // Clear local buffer after transferring
+              });
             }
-        })
-    }
-
-    public getStateObservable(): BehaviorSubject<ConnectionState> {
-        return this.connectionState;
-    }
-
-    private async transferLocalBufferToMongoDB(): Promise<void> {
-        return new Promise((resolve, reject) => {
-            console.log(`Releasing buffer Message: currently there is ${this.messageBuffer.length}. Transferring to database...`)
-            if (this.messageModel) {
-                this.messageBuffer.forEach(async message => {
-                    try {
-                        if (this.messageModel) {
-                            await this.messageModel.create(message);
-                        }
-                    } catch (error) {
-                        console.error('Error transferring message to MongoDB:', error);
-                    }
-                })
-                this.messageBuffer = []; // Clear local buffer after transferring
-            }
-        })
-    }
+          },
+        });
+      }
+    });
+  }
 
-    // Additional methods as required...
+  // Additional methods as required...
 }

services/fis.retransmission.service.ts

@@ -1,461 +0,0 @@
-import mongoose, { Model, Schema } from 'mongoose';
-import { BehaviorSubject, Observable, Subject, Subscription, from } from 'rxjs'
-import { ColorCode, Message, MessageLog, ReportStatus, Status } from '../interfaces/general.interface'
-import { resolve } from 'path';
-require('dotenv').config();
-
-// Implement status chain refactoring
-export class FisRetransmissionService {
-    private mongoUrl: string = process.env.MONGO as string
-    private bufferedStorage: Message[] = []
-    private mongoConnection: any
-    private messageModel: Model<any> | null | undefined
-    private maximumBufferLength: number = parseInt(process.env.MaxBufferLoad as string) // please configure at environment
-
-    constructor(private databaseName: string, private statusReport: BehaviorSubject<ReportStatus>) {
-        // Connect to mongoDB. 
-        this.manageMongoConnection(databaseName)
-    }
-
-    // Main function that intercepts outgoing messages by communicating || intepreting report status from grpc connection as indicator 
-    public handleMessage(applicationOutgoingMessage: Subject<Message>): Subject<Message> {
-        let releaseMessageSubject: Subject<Message> = new Subject() // Every message subscribed from applicationOutgoingMessage will be released through this subject
-        let messageReleaseSubscription: Subscription | null = null
-        let messageBufferSubscription: Subscription | null = null
-        let messageStreamToMongo: Subscription | null = null
-        this.checkBufferLimit(applicationOutgoingMessage, this.statusReport)
-        this.statusReport.subscribe((report: ReportStatus) => {
-            /* Green should release all data from buffer and mongo and also redirect the applicationOutgoingMessage back into the return subject(releaseMessageSubject)
-            if there's any. */
-            if (report.code == ColorCode.GREEN) {
-                let status: Status = 1
-                // console.log(`Connection status report && ${report.message ?? 'No Message'}`)
-                /* Status Chain begins */
-                if (status === 1) {
-                    messageStreamToMongo = this.deactivateMongoStreamSubscription(messageStreamToMongo)
-                    if (messageStreamToMongo) status = 0
-                }
-                if (status === 1) {
-                    messageBufferSubscription = this.deactivateBufferSubscription(messageBufferSubscription)
-                    if (messageBufferSubscription) status = 0
-                }
-                if (status === 1) {
-                    messageReleaseSubscription = this.activateReleaseSubscription(messageReleaseSubscription, applicationOutgoingMessage, releaseMessageSubject)
-                    if (!messageReleaseSubscription) status = 0
-                }
-                if (status === 1) {
-                    this.releaseMessageFromLocalBuffer(this.bufferedStorage).then((resObs: Observable<Message>) => {
-                        resObs.subscribe({
-                            next: message => releaseMessageSubject.next(message),
-                            error: err => console.error(err),
-                            complete: () => {
-                                this.bufferedStorage = []
-                                console.log(`Reset buffer Storage count: ${this.bufferedStorage.length}. All messages have been released back into the stream.`)
-                            }
-                        })
-                    }).catch((err) => {
-                        status = 0
-                        console.error(err)
-                    })
-                }
-                if (status === 1) {
-                    this.releaseMessageFromMongoStorage().then((resObs: Subject<Message>) => {
-                        resObs.subscribe({
-                            next: message => releaseMessageSubject.next(message),
-                            error: err => console.error(err),
-                            complete: () => console.log(`All Mongo data are transferred `)
-                        })
-                    }).catch((err) => {
-                        status = 0
-                        console.error(err)
-                    })
-                }
-                if (status === 0) {
-                    console.log(`Something Went Wrong in handling ${ColorCode.RED} report.`)
-                }
-            }
-            /* Start buffering the messages coming in from applicationOutgonigMessages and also stop it from flowing into the release subject */
-            if (report.code == ColorCode.YELLOW) {
-                if (report.payload) {
-                    console.log(`Rebuffering ${report.payload.message?.appData?.msgId} into buffer...`)
-                    this.bufferedStorage.push(report.payload)
-                }
-                // console.log(`Connection status report && ${report.message ?? 'No Message'}`)
-                let status: Status = 1
-                /* Status Chain begins */
-                if (status === 1) {
-                    messageBufferSubscription = this.activateBufferSubscription(this.bufferedStorage, messageBufferSubscription, applicationOutgoingMessage)
-                    if (!messageBufferSubscription) status = 0
-                }
-                if (status === 1) {
-                    messageReleaseSubscription = this.deactivateReleaseSubscription(messageReleaseSubscription)
-                    if (messageReleaseSubscription) status = 0
-                }
-                if (status === 0) {
-                    console.log(`Something Went Wrong in handling ${ColorCode.RED} report.`)
-                }
-            }
-            /* Stop buffering the message in local instance, but start saving them in database. Must first transfer the ones in local buffer before redirecting the 
-            flow from applicationOutgoingMessage into Mongo */
-            if (report.code == ColorCode.RED) {
-                // console.log(`Connection status report: ${report.message}`)
-                if (report.payload) {
-                    console.log(`Rebuffering ${report.payload.message?.appData?.msgId} into storage...`)
-                    this.saveToMongo(report.payload)
-                }
-                console.log(`Connection status report && ${report.message ?? 'No Message'}`)
-                let status: Status = 1
-                if (status === 1) {
-                    messageStreamToMongo = this.activateMongoStreamSubscription(messageStreamToMongo, applicationOutgoingMessage)
-                    if (!messageStreamToMongo) status = 0
-                }
-                if (status === 1) {
-                    messageReleaseSubscription = this.deactivateReleaseSubscription(messageReleaseSubscription)
-                    if (messageReleaseSubscription) status = 0
-                }
-                if (status === 1) {
-                    messageBufferSubscription = this.deactivateBufferSubscription(messageBufferSubscription)
-                    if (messageBufferSubscription) status = 0
-                }
-                if (status === 1) {
-                    this.transferBufferedMessageToMongoStorage(this.bufferedStorage, messageBufferSubscription).then((res: any[]) => {
-                        if (res.length !== this.bufferedStorage.length || this.bufferedStorage.length > 0) status = -1 // this promise function should return an empty array
-                    })
-                }
-                if (status === 0) {
-                    console.log(`Something Went Wrong in handling ${ColorCode.RED} report.`)
-                }
-            }
-            if (!report.code) {
-                console.log(`Unknown message...`)
-            }
-        })
-        return releaseMessageSubject
-    }
-
-    private checkIfMongoBufferHasData(): Promise<boolean> {
-        return new Promise(async (resolve, reject) => {
-            if (this.messageModel) {
-                try {
-                    const count = await this.messageModel.countDocuments();
-                    resolve(count > 0)
-                }
-                catch (error) {
-                    reject(error)
-                }
-            }
-        })
-    }
-
-    // IF Buffer exceeds a certain limit, it will trigger RED. Configure in .env file. There's the concern of 2 RED status, one from this and another from other means.
-    // Behaviour of this needs to be investigated further
-    private checkBufferLimit(message: Subject<Message>, statusReport: Subject<ReportStatus>) {
-        let status: Status = 1
-        if (status = 1) {
-            message.subscribe(() => {
-                if (this.bufferedStorage.length >= this.maximumBufferLength) {
-                    // for every messges that comes in, check the bufffer size, if it exceesd more than designated amount, push a red report status i
-                    console.log(`Buffer length exceeds limit imposed!!!`)
-                    let report: ReportStatus = {
-                        code: ColorCode.RED,
-                        message: `Buffer is exceeding limit. Initiate storage transfer to designated database.`,
-                    }
-                    statusReport.next(report)
-                }
-            })
-        }
-    }
-
-    // Release the incoming Messages to be returned to the caller
-    private activateReleaseSubscription(messageReleaseSubscription: Subscription | null, applicationOutgoingMessage: Subject<Message>, releaseMessageSubject: Subject<Message>): Subscription | null {
-        let status: Status = 1
-        if (status = 1) {
-            if (!messageReleaseSubscription) {
-                messageReleaseSubscription = applicationOutgoingMessage.subscribe({
-                    next: (message: Message) => {
-                        console.log(`Releasing ${(message.message as MessageLog).appData.msgId}...`);
-                        releaseMessageSubject.next(message);
-                    },
-                    error: (err) => console.error(err),
-                    complete: () => { },
-                });
-                console.log(`Subscription message release activated.`);
-            } else {
-                status = 0
-                console.log(`Subscription message release is already active.`);
-            }
-        }
-        return messageReleaseSubscription
-    }
-
-    // Stop the incoming Messages to be returned to caller
-    private deactivateReleaseSubscription(messageReleaseSubscription: Subscription | null): Subscription | null {
-        let status: Status = 1
-        if (status = 1) {
-            if (messageReleaseSubscription) {
-                messageReleaseSubscription.unsubscribe();
-                messageReleaseSubscription = null;
-                console.log(`Subscription message release deactivated.`);
-            } else {
-                console.log(`Subscription message release is already deactivated.`);
-            }
-        }
-        return messageReleaseSubscription
-    }
-
-    // Begin to push the incoming messages into local instantarray
-    private activateBufferSubscription(bufferStorage: Message[], messageBufferSubscription: Subscription | null, applicationOutgoingMessage: Subject<Message>): Subscription | null {
-        let status: Status = 1
-        if (status = 1) {
-            if (!messageBufferSubscription) {
-                messageBufferSubscription = applicationOutgoingMessage.subscribe({
-                    next: (message: any) => {
-                        console.log(`Buffering ${(message.message as MessageLog).appData.msgId}...  Local array length: ${bufferStorage.length}`);
-                        bufferStorage.push(message)
-                    },
-                    error: (err) => console.error(err),
-                    complete: () => { },
-                });
-                console.log(`Subscription message buffer activated.`);
-            } else {
-                status = 0
-                console.log(`Subscription message buffer is already active.`);
-            }
-        }
-        return messageBufferSubscription
-    }
-
-    // Stop pushing the incoming messages into local instantarray
-    private deactivateBufferSubscription(messageBufferSubscription: Subscription | null): Subscription | null {
-        let status: Status = 1
-        if (status) {
-
-            if (messageBufferSubscription) {
-                messageBufferSubscription.unsubscribe();
-                messageBufferSubscription = null;
-                console.log(`Subscription message buffer deactivated.`);
-            } else {
-                status = 0
-                console.log(`Subscription message buffer is already deactivated.`);
-            }
-        }
-        return null
-    }
-
-    // Change the streaming direction of the incoming messages into mongo streaming subject( to be saved in local databse )
-    private activateMongoStreamSubscription(messageStreamToMongo: Subscription | null, applicationOutgoingMessage: Subject<Message>): Subscription | null {
-        let status: Status = 1
-        if (status = 1) {
-            if (!messageStreamToMongo) {
-                messageStreamToMongo = applicationOutgoingMessage.subscribe({
-                    next: (message: any) => {
-                        console.log(`Saving ${(message.message as MessageLog).appData.msgId}...`);
-                        this.saveToMongo(message)
-                    },
-                    error: (err) => console.error(err),
-                    complete: () => { },
-                });
-                console.log(`Subscription message streaming to Mongo activated.`);
-            } else {
-                status = 0
-                console.log(`Subscription message streaming to Mongo  is already active.`);
-            }
-        }
-        return messageStreamToMongo
-    }
-
-    // Stop or cut off the mongo streaming
-    private deactivateMongoStreamSubscription(messageStreamToMongo: Subscription | null): Subscription | null {
-        let status: Status = 1
-        if (status = 1) {
-            if (messageStreamToMongo) {
-                messageStreamToMongo.unsubscribe();
-                messageStreamToMongo = null;
-                console.log(`Subscription message streaming to Mongo deactivated.`);
-            } else {
-                status = 0
-                console.log(`Subscription message streaming to Mongo is already deactivated.`);
-            }
-        }
-        return messageStreamToMongo
-    }
-
-    // To be used by mongoStreamSubscription to perform the saving execution
-    private async saveToMongo(message: Message): Promise<boolean> {
-        return new Promise((resolve, reject) => {
-            // let messageModel: Model<any> = this.mongoConnection.model('Message', require('../models/message.schema'))
-            if (this.messageModel) {
-                this.messageModel.create(message).then(() => {
-                    console.log(`Saved MessageID ${(message.message as MessageLog).appData.msgId} into ${this.mongoUrl}`);
-                    resolve(true)
-                }).catch((err) => {
-                    console.log(`MongoSaveError: ${err.message}`)
-                    reject(err)
-                })
-            } else {
-                console.log(`Cant save message. Message Model is absent or not properly initialized`)
-            }
-        })
-    }
-
-    // As the name implies, transder all the messages from the local instance into mongoStorage. Local instance should be emptied after transfer is completed
-    private async transferBufferedMessageToMongoStorage(bufferedMessage: Message[], messageBufferSubscription: Subscription | null): Promise<Message[]> {
-        return new Promise((resolve, reject) => {
-            let status: Status = 1
-            if (status = 1) {
-                let bufferedStorage: Observable<Message> = from(bufferedMessage)
-                bufferedStorage.subscribe({
-                    next: (message: any) => {
-                        this.saveToMongo(message).then((res) => {
-                            console.log(`Message ${(message.message as MessageLog).appData.msgId} saved successfully...`)
-                        }).catch((err) => console.error(err))
-                    },
-                    error: (error) => {
-                        reject(error)
-                        console.error(error)
-                    },
-                    complete: () => {
-                        this.bufferedStorage = []
-                        if (messageBufferSubscription) {
-                            console.log(`All ${bufferedMessage.length} buffered messages have been sent for transfer to ${this.mongoUrl}. Current length: ${this.bufferedStorage.length}`)
-                        }
-                        resolve(this.bufferedStorage)
-                    }
-                })
-            }
-        })
-    }
-
-    // Transfer stored messages from the local instance back into the stream to be returned to the caller.
-    private async releaseMessageFromLocalBuffer(bufferedStorage: Message[]): Promise<Observable<Message>> {
-        return new Promise((resolve, reject) => {
-            let status: Status = 1
-            if (status = 1) {
-                if (bufferedStorage.length > 1) {
-                    let caseVariable = this.bufferedStorage.length > 1;
-                    console.log(`Releasing data from local buffer instance. There ${caseVariable ? "is" : "are"} ${this.bufferedStorage.length} messages...`);
-                    let returnArrayObs: Observable<Message> = from(bufferedStorage)
-                    resolve(returnArrayObs)
-                } else {
-                    let message = `There is no data in stored in local instance`
-                    reject(message)
-                }
-            }
-        })
-    }
-
-    // Transder all the stored messages in designated mongo databases. It should be empty after all the data has been transferred.
-    private async releaseMessageFromMongoStorage(): Promise<Subject<Message>> {
-        return new Promise((resolve, reject) => {
-            let status: Status = 1
-            if (status = 1) {
-                let dataSubject: Subject<Message> = new Subject()
-                this.extractAllMessages(dataSubject)
-                resolve(dataSubject)
-            }
-        })
-    }
-
-    // Connect to designated mongodatabase.
-    private async connectToMongoDatabase(databaseName: string): Promise<any> {
-        return new Promise((resolve, reject) => {
-            let status: Status = 1
-            if (status = 1) {
-                let database = this.mongoUrl + databaseName
-                console.log(`Connected to ${database}`)
-                this.mongoConnection = mongoose.createConnection(database)
-                this.mongoConnection.on('error', (error) => {
-                    console.error('Connection error:', error);
-                    resolve('')
-                });
-                this.mongoConnection.once('open', () => {
-                    // console.log(`Connected to ${process.env.MONGO}`);
-                    let report: ReportStatus = {
-                        code: ColorCode.RED,
-                        message: `Mongo storage available`
-                    }
-                    this.messageModel = this.mongoConnection.model('Message', require('../models/message.schema'));
-                    this.statusReport.next(report)
-                });
-            }
-        })
-    }
-
-    // Manage mongoCOnnectino. The logic used would be different across differnet application. This will loop the process indefinitely os it is always trying to connect to database.
-    private async manageMongoConnection(databaseName: string): Promise<boolean> {
-        while (true) {
-            try {
-                await this.connectToMongoDatabase(databaseName)
-            } catch (error) {
-                console.log(`Something Wrong occured. Please check at manageMongoConnection`)
-            }
-            await new Promise(resolve => setTimeout(resolve, 1000)); // Wait for 1 second before the next attempt
-        }
-    }
-
-    // This will be used to release all the hostage messages once the light is green.
-    public async extractAllMessages(subjectArgs: Subject<Message>): Promise<void> {
-        // Need to resolve the issue of streaming in a specific order that is sequential
-        let status: Status = 1
-        if (status = 1) {
-            if (this.messageModel) {
-                const eventStream = this.messageModel.find().lean().cursor()
-                eventStream.on('data', (message) => {
-                    // Emit each document to the subject
-                    subjectArgs.next(message);
-                });
-                eventStream.on('end', async () => {
-                    // All data has been streamed, complete the subject
-                    subjectArgs.complete();
-                    // Delete the data once it has been streamed
-                    try {
-                        if (this.messageModel) {
-                            await this.messageModel.deleteMany({});
-                            console.log('Data in Mongo deleted successfully.');
-                        } else {
-                            console.log(`Message Mongoose Model is not intiated properly...`)
-                        }
-                    } catch (err) {
-                        console.error('Error deleting data:', err);
-                    }
-                });
-            } else {
-                status = 0
-                console.log(`Error: Message Model is ${this.messageModel}!! Please set up the mongoose connection properly!`)
-            }
-        }
-    }
-
-}
-
-
-
-
-
-// Store in json file in this project folder. To be enabled in future
-// private async transferMessageToLocalStorage(message: Subject<any>): Promise<void> {
-//     let localArray: any[] = this.bufferedStorage
-//     let filename = `localstorage.json`;
-
-//     while (localArray.length > 0) {
-//         let objectToWrite = this.bufferedStorage[0];
-//         await writeMessage(objectToWrite, filename)
-//     }
-//     message.subscribe((message: any) => {
-//         writeMessage(message, filename)
-//     })
-
-//     if (localArray.length < 1) this.bufferedStorage = localArray
-//     console.log('Local Array is empty. Finished transferring to files.')
-
-//     async function writeMessage(message: any, filename: string) {
-//         try {
-//             let stringifiedMessage = JSON.stringify(message);
-//             await fs.promises.appendFile(filename, stringifiedMessage + "\r\n")
-//             console.log(`Successfully transferred ${filename}`);
-//             localArray.shift();
-//         } catch (err) {
-//             console.error(`Error trasferring ${filename}:`, err);
-//         }
-//     }
-// }

services/grpc.service.method.ts

@@ -1,49 +1,31 @@
 import * as grpc from '@grpc/grpc-js';
 import { Subject, Subscription } from "rxjs";
-import { ReportStatus, ColorCode, Message, MessageLog, ConnectionAttribute, ConnectionRequest, GrpcConnectionType, ConnectionState } from "../interfaces/general.interface";
+import { Message, ConnectionAttribute, ConnectionRequest, GrpcConnectionType, ConnectionState, MessageLog, State, OutGoingInfo } from "../interfaces/general.interface";
 import { Status } from '@grpc/grpc-js/build/src/constants';
-import { v4 as uuidv4 } from 'uuid'
 import { message_proto } from './protos/server.proto'
 import { ServerWritableStreamImpl } from '@grpc/grpc-js/build/src/server-call';
 export class GrpcServiceMethod {
     private server: grpc.Server | any
     private messageToBeSendOver: Message | any
-    private callRequestsFromRemote: ServerWritableStreamImpl<any, ResponseType>[] = []
+    private clientInfo: any[] = []
+    // private callRequestsFromRemote: ServerWritableStreamImpl<any, ResponseType>[] = []
 
-    public async create(request: ConnectionRequest, connectionAttribute: ConnectionAttribute): Promise<any> {
+    public async create(request: ConnectionRequest, connectionAttribute: ConnectionAttribute, outGoingInfo: OutGoingInfo): Promise<any> {
         // Assuming currently only one client
-        this.createGrpcInstance(request.server.serverUrl, { instanceType: 'server' }, connectionAttribute)
-        this.createGrpcInstance(request.client.targetServer, { instanceType: 'client' }, connectionAttribute)
-    }
-
-    // For testing only
-    public async shutDownServer(): Promise<string> {
-        return new Promise((resolve, reject) => {
-            console.log(`Shutting down servers...`)
-            if (this.server) {
-                this.callRequestsFromRemote[0].destroy()
-                this.callRequestsFromRemote[0].end()
-                this.server.forceShutdown()
-                let message: string = `Server shut down successfully!`
-                resolve(message)
-            }
-            if (!this.server) {
-                let errorMsg: string = `There's no active server here`
-                reject(errorMsg)
-            }
-        })
+        this.createGrpcInstance(request.server.serverUrl, { instanceType: 'server' }, connectionAttribute, outGoingInfo)
+        this.createGrpcInstance(request.client.targetServer, { instanceType: 'client' }, connectionAttribute, outGoingInfo)
     }
 
     private async generateAdditionalAttributes(connectionAttribute: ConnectionAttribute, clientInfo?: any, localInfo?: any) {
         if (clientInfo) {
-            connectionAttribute.inComing.StreamID = clientInfo.channelID
-            connectionAttribute.inComing.PublisherID = clientInfo.publisherID
-            connectionAttribute.inComing.SubscriberID = clientInfo.subscriberID
+            connectionAttribute.inComing.StreamID = clientInfo.StreamID
+            connectionAttribute.inComing.PublisherID = clientInfo.PublisherID
+            connectionAttribute.inComing.SubscriberID = clientInfo.SubscriberID
         }
         if (localInfo) {
-            connectionAttribute.outGoing.StreamID = localInfo.channelID
-            connectionAttribute.outGoing.PublisherID = localInfo.publisherID
-            connectionAttribute.outGoing.SubscriberID = localInfo.subscriberID
+            connectionAttribute.outGoing.StreamID = localInfo.StreamID
+            connectionAttribute.outGoing.PublisherID = localInfo.PublisherID
+            connectionAttribute.outGoing.SubscriberID = localInfo.SubscriberID
         }
         if (connectionAttribute.outGoing.StreamID && connectionAttribute.inComing.StreamID) {
             connectionAttribute.ConnectionID.local = connectionAttribute.outGoing.StreamID + connectionAttribute.inComing.StreamID
@@ -51,18 +33,12 @@ export class GrpcServiceMethod {
         }
     }
 
-    // To be migrated into a service in the immediate future
     private async createGrpcInstance(
         serverUrl: string,
         grpcType: GrpcConnectionType,
         connectionAttribute: ConnectionAttribute,
+        outGoingInfo: OutGoingInfo
     ) {
-        let statusControl: Subject<ConnectionState> = connectionAttribute.connectionStatus
-        let consecutiveResolutions = 0;
-        let lastResolutionTime = Date.now();
-        let yellowErrorEmission: boolean = false
-        let redErrorEmission: boolean = false
-
         while (true) {
             try {
                 let recreatePromise = new Promise((resolve) => {
@@ -72,40 +48,15 @@ export class GrpcServiceMethod {
                         })
                     }
                     if (grpcType.instanceType == 'client') {
-                        this.createServerStreamingClient(serverUrl, connectionAttribute).then(() => {
+                        this.createServerStreamingClient(serverUrl, connectionAttribute, outGoingInfo).then(() => {
                             resolve('recreate')
                         })
                     }
                 })
                 await recreatePromise
-                // If connection resolves (indicating failure), increment the count
-                consecutiveResolutions++;
-                // console.log(`Reconnection Attempt: ${consecutiveResolutions}`)
-                if (redErrorEmission == false) {
-                    redErrorEmission = true
-                    console.error(`Connection failed ${consecutiveResolutions} times. Stopping connection attempts.`);
-                    let error: ConnectionState = {
-                        status: 'BUFFER',
-                        reason: `Server is not responding...`
-                    }
-                    statusControl.next(error)
-                }
-
             } catch (error) {
-                // Connection did not resolve, reset the count
-                consecutiveResolutions = 0;
                 console.error('Connection attempt failed:', error);
             }
-            // Check for a pause of more than 2 seconds since the last resolution attempt
-            const currentTime = Date.now();
-            const timeSinceLastResolution = currentTime - lastResolutionTime;
-            if (timeSinceLastResolution > 2000) {
-                consecutiveResolutions = 0;
-                yellowErrorEmission = false
-                redErrorEmission = false
-            }
-            // Update the last resolution time
-            lastResolutionTime = currentTime;
             await new Promise(resolve => setTimeout(resolve, 1000)); // Wait for 1 second before the next attempt
             // timeout generate message to trigger this reconnection
         }
@@ -119,22 +70,23 @@ export class GrpcServiceMethod {
         return new Promise((resolve, reject) => {
             try {
                 if (!this.server) {
-                    this.server = new grpc.Server();
+                    this.server = new grpc.Server()
+                } else {
+                    console.log(`Grpc server alrady started.`) // this kept calling, that means this function is resolving on it's own, prompting the reconnection logic
                 }
 
                 this.server.addService(message_proto.Message.service, {
                     HandleMessage: (call) => {
-                        this.callRequestsFromRemote.push(call)
                         let clientInfo = JSON.parse(call.request.message)
-                        this.generateAdditionalAttributes(connectionAttribute, clientInfo)
+                        this.clientInfo.push(clientInfo)
+                        // this.generateAdditionalAttributes(connectionAttribute, clientInfo)
 
                         console.log(`Initializing stream. Opening Channel... Confirmation from ${call.request.id}`)
 
                         if (connectionAttribute.outGoing.MessageToBePublished) {
                             let subscription: Subscription = connectionAttribute.outGoing.MessageToBePublished.subscribe({
                                 next: (response: Message) => {
-                                    // console.log(`Sending from GRPC server: ${(response.message as MessageLog).appData.msgId} `)
-                                    this.messageToBeSendOver = response
+                                    console.log(`Sending from GRPC server: ${(response.message as MessageLog).appData.msgId} `)
                                     let message = {
                                         id: response.id,
                                         message: JSON.stringify(response.message)
@@ -152,16 +104,13 @@ export class GrpcServiceMethod {
                                     resolve('')
                                 }
                             })
+                            console.log(connectionAttribute)
+                            let report: ConnectionState = {
+                                status: 'DIRECT_PUBLISH'
+                            }
+                            connectionAttribute.connectionStatus.next(report)
                         }
-
-                        console.log(connectionAttribute)
-                        let report: ConnectionState = {
-                            status: 'DIRECT_PUBLISH'
-                        }
-                        connectionAttribute.connectionStatus.next(report)
-
                     },
-
                     Check: (_, callback) => {
                         // for now it is just sending the status message over to tell the client it is alive
                         // For simplicity, always return "SERVING" as status
@@ -177,23 +126,17 @@ export class GrpcServiceMethod {
             catch (error) {
                 resolve(error)
             }
-
         })
     }
 
     // Send a request over to the other server to open a channel for this server to emit/stream messages over
     public async createServerStreamingClient(
         server: string,
-        connectionAttribute: ConnectionAttribute
+        connectionAttribute: ConnectionAttribute,
+        outGoingInfo: OutGoingInfo
     ): Promise<string> {
         return new Promise(async (resolve, reject) => {
             const client = new message_proto.Message(server, grpc.credentials.createInsecure());
-
-            let outGoingInfo: any = {
-                channelID: uuidv4(),
-                publisherID: uuidv4(),
-                subscriberID: uuidv4()
-            }
             this.generateAdditionalAttributes(connectionAttribute, {}, outGoingInfo)
 
             let call = client.HandleMessage({ id: server, message: JSON.stringify(outGoingInfo) })
@@ -233,7 +176,7 @@ export class GrpcServiceMethod {
 
 
     // THis is no longer necesarry after the introduction of connection Attribute. But it is still useful for checking for the other side's health
-    public async checkConnectionHealth(client: any, statusControl: Subject<ReportStatus>, alreadyHealthCheck: boolean): Promise<boolean> {
+    public async checkConnectionHealth(client: any, statusControl: Subject<ConnectionState>, alreadyHealthCheck: boolean): Promise<boolean> {
         return new Promise((resolve, reject) => {
             client.Check({}, (error, response) => {
                 if (response) {

services/server-client.service.ts

@@ -1,8 +1,8 @@
-import { BehaviorSubject, Subject } from 'rxjs';
-import { ColorCode, Message, ReportStatus, ConnectionAttribute, ConnectionRequest, ConnectionState, MessageLog } from '../interfaces/general.interface';
+import { BehaviorSubject } from 'rxjs';
+import { ConnectionAttribute, ConnectionRequest, ConnectionState, OutGoingInfo } from '../interfaces/general.interface';
 import { GrpcServiceMethod } from './grpc.service.method';
-import { FisRetransmissionService } from './fis.retransmission.service';
 import { BufferService } from './buffer.service';
+import { v4 as uuidv4 } from 'uuid'
 import * as dotenv from 'dotenv'
 dotenv.config()
 
@@ -10,22 +10,17 @@ export class ServerClientManager {
 
     private connectionAttributes: ConnectionAttribute[] = []
     private request: ConnectionRequest | any
+    private outGoingInfo: OutGoingInfo
+    private grpcService: GrpcServiceMethod = new GrpcServiceMethod()
 
-    constructor(private grpcService: GrpcServiceMethod) {
+    constructor() {
+        this.outGoingInfo = {
+            StreamID: uuidv4(),
+            PublisherID: uuidv4(),
+            SubscriberID: uuidv4()
+        }
     }
 
-    // putting it here for later use. to be used for testing
-    public restartServerInDuration(time: number) {
-        console.log(this.request)
-        console.log(this.connectionAttributes)
-        this.grpcService.shutDownServer().then((msg: string) => {
-            console.log(msg)
-            setTimeout(() => {
-                this.generateConnection(this.request)
-            }, time * 1000)
-        })
-
-    }
 
     public generateConnection(request: ConnectionRequest) {
         this.request = request
@@ -39,10 +34,6 @@ export class ServerClientManager {
         let initialReport: ConnectionState = { status: 'BUFFER' }
         let reportSubject: BehaviorSubject<ConnectionState> = new BehaviorSubject(initialReport)
         let retransmission: BufferService = new BufferService(request.server.messageToBePublishedFromApplication, reportSubject, database)
-        // let initialReport: ReportStatus = { code: ColorCode.RED, message: 'Initialization of the subject' }
-        // let reportSubject: BehaviorSubject<ReportStatus> = new BehaviorSubject(initialReport)
-        // let retransmission: FisRetransmissionService = new FisRetransmissionService(database, reportSubject)
-        // let messageToBePublished: Subject<Message> = retransmission.handleMessage(request.server.messageToBePublishedFromApplication) ?? request.server.messageToBePublishedFromApplication
 
         let connectionAttribute: ConnectionAttribute = {
             ConnectionID: {
@@ -60,14 +51,16 @@ export class ServerClientManager {
             connectionStatus: reportSubject
         }
 
-        // This is default connection
+        // connectionAttribute.outGoing.MessageToBePublished?.subscribe(e => console.log((e.message as MessageLog).appData.msgId))
+        // This is default connection`
         if (!request.server.connectionType) {
             request.server.connectionType = 'GRPC'
         }
         // For each connection type:
         if (request.server.connectionType == 'GRPC') {
-            this.grpcService.create(request, connectionAttribute)
+            this.grpcService.create(request, connectionAttribute, this.outGoingInfo)
             this.connectionAttributes.push(connectionAttribute)
+            console.log(`There is now ${this.connectionAttributes.length} connection Attributes`)
         }
     }
 

test/grpc1.ts

@@ -9,7 +9,7 @@ import mongoose from 'mongoose';
 mongoose.connect('mongodb://localhost:27017/grpc1')
 const Message = mongoose.model('Message', require('../models/message.schema'))
 // Subject for bidirectional communication
-const connectionService: ServerClientManager = new ServerClientManager(new GrpcServiceMethod())
+const connectionService: ServerClientManager = new ServerClientManager()
 const messagesJSON: any = readFileSync('payload.json')
 let parsedMessages: any[] = JSON.parse(messagesJSON) // load the fake messages generated for this trial 
 let targetserver: string = 'localhost:3001'
@@ -20,26 +20,26 @@ let intervalToStreamOutGoingMessage: number = 1
 
 
 /* Simple Test: 1 to 1 */
-let connectionRequest: ConnectionRequest = {
-  server: {
-    name: 'g1',
-    serverUrl: hostServer,
-    connectionType: 'GRPC',
-    messageToBePublishedFromApplication: new Subject<Message>()
-  },
-  client: {
-    name: 'g2',
-    targetServer: targetserver,
-    connectionType: 'GRPC',
-    messageToBeReceivedFromRemote: new Subject<Message>()
-  }
-}
+// let connectionRequest: ConnectionRequest = {
+//   server: {
+//     name: 'g1',
+//     serverUrl: hostServer,
+//     connectionType: 'GRPC',
+//     messageToBePublishedFromApplication: new Subject<Message>()
+//   },
+//   client: {
+//     name: 'g2',
+//     targetServer: targetserver,
+//     connectionType: 'GRPC',
+//     messageToBeReceivedFromRemote: new Subject<Message>()
+//   }
+// }
 
-connectionService.generateConnection(connectionRequest)
+// connectionService.generateConnection(connectionRequest)
 
 // let generateFakeMessagesToBePublished = stream().pipe(take(1000))
 
-// let generateFakeMessagesToBePublished = from(parsedMessages).pipe(take(10000))
+// let generateFakeMessagesToBePublished = from(parsedMessages).pipe(take(5000))
 // generateFakeMessagesToBePublished.subscribe({
 //   next: message => {
 //     let payload: Message = {
@@ -52,25 +52,25 @@ connectionService.generateConnection(connectionRequest)
 //   complete: () => console.log(`Completed`)
 // })
 
-stream().subscribe({
-  next: message => {
-    let payload = {
-      id: hostServer,
-      message: message
-    }
-    connectionRequest.server.messageToBePublishedFromApplication.next(payload)
-  }
-})
+// stream().subscribe({
+//   next: message => {
+//     let payload = {
+//       id: hostServer,
+//       message: message
+//     }
+//     connectionRequest.server.messageToBePublishedFromApplication.next(payload)
+//   }
+// })
 
-connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
-  next: response => {
-    console.log(`Received ${(response.message as MessageLog).appData.msgId} from ${connectionRequest.client.targetServer}`)
-    // Message.create(response)
-    array.push(response)
-  },
-  error: error => console.error(error),
-  complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
-})
+// connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
+//   next: response => {
+//     console.log(`Received ${(response.message as MessageLog).appData.msgId} from ${connectionRequest.client.targetServer}`)
+//     // Message.create(response)
+//     array.push(response)
+//   },
+//   error: error => console.error(error),
+//   complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
+// })
 
 
 
@@ -134,37 +134,52 @@ connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
 
 
 /* Simple Test: 1 to Many */
-// let connectionRequest: ConnectionRequest = {
-//   server: {
-//     name: 'g1',
-//     serverUrl: hostServer,
-//     connectionType: 'GRPC',
-//     messageToBePublishedfromApplication: new Subject<Message>()
-//   },
-//   client: {
-//     name: 'g2',
-//     targetServer: targetserver,
-//     connectionType: 'GRPC',
-//     messageToBeReceivedFromRemote: new Subject<Message>()
-//   }
-// }
-// let connectionRequest2: ConnectionRequest = {
-//   server: {
-//     name: 'g1',
-//     serverUrl: hostServer,
-//     connectionType: 'GRPC',
-//     messageToBePublishedfromApplication: new Subject<Message>()
-//   },
-//   client: {
-//     name: 'g3',
-//     targetServer: targetserver2,
-//     connectionType: 'GRPC',
-//     messageToBeReceivedFromRemote: new Subject<Message>()
-//   }
-// }
+let connectionRequest: ConnectionRequest = {
+  server: {
+    name: 'g1',
+    serverUrl: hostServer,
+    connectionType: 'GRPC',
+    messageToBePublishedFromApplication: new Subject<Message>()
+  },
+  client: {
+    name: 'g2',
+    targetServer: targetserver,
+    connectionType: 'GRPC',
+    messageToBeReceivedFromRemote: new Subject<Message>()
+  }
+}
+let connectionRequest2: ConnectionRequest = {
+  server: {
+    name: 'g1',
+    serverUrl: hostServer,
+    connectionType: 'GRPC',
+    messageToBePublishedFromApplication: new Subject<Message>()
+  },
+  client: {
+    name: 'g3',
+    targetServer: targetserver2,
+    connectionType: 'GRPC',
+    messageToBeReceivedFromRemote: new Subject<Message>()
+  }
+}
 
-// connectionService.generateConnection(connectionRequest)
-// connectionService.generateConnection(connectionRequest2)
+
+connectionService.generateConnection(connectionRequest)
+connectionService.generateConnection(connectionRequest2)
+
+let generateFakeMessagesToBePublished = from(parsedMessages).pipe(take(50))
+generateFakeMessagesToBePublished.subscribe({
+  next: message => {
+    let payload: Message = {
+      id: hostServer,
+      message: message
+    }
+    connectionRequest.server.messageToBePublishedFromApplication.next(payload)
+    connectionRequest2.server.messageToBePublishedFromApplication.next(payload)
+  },
+  error: error => console.error(error),
+  complete: () => console.log(`Completed`)
+})
 
 // let generateFakeMessagesToBePublished = stream().pipe(take(10))
 // generateFakeMessagesToBePublished.subscribe({
@@ -173,35 +188,28 @@ connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
 //       id: hostServer,
 //       message: message
 //     }
-//     connectionRequest.server.messageToBePublishedfromApplication.next(payload)
+//     connectionRequest.server.messageToBePublishedFromApplication.next(payload)
+//     connectionRequest2.server.messageToBePublishedFromApplication.next(payload)
 //   }
 // })
-// connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
-//   next: request => {
-//     console.log(`Received ${(response.message as MessageLog).appData.msgId} from ${connectionRequest.client.targetServer}`)
 
-//     array.push(request)
-//   },
-//   error: error => console.error(error),
-//   complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
-// })
-
-// connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
-//   next: request => {
-// console.log(`Received ${(response.message as MessageLog).appData.msgId} from ${connectionRequest.client.targetServer}`)
-//     array.push(request)
-//   },
-//   error: error => console.error(error),
-//   complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
-// })
+connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
+  next: request => {
+    console.log(`Received ${(request.message as MessageLog).appData.msgId} from ${connectionRequest.client.targetServer}`)
+    array.push(request)
+  },
+  error: error => console.error(error),
+  complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
+})
 
-// connectionRequest2.client.messageToBeReceivedFromRemote.subscribe({
-//   next: request => {
-//     array.push(request)
-//   },
-//   error: error => console.error(error),
-//   complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
-// })
+connectionRequest2.client.messageToBeReceivedFromRemote.subscribe({
+  next: request => {
+    console.log(`Received ${(request.message as MessageLog).appData.msgId} from ${connectionRequest.client.targetServer}`)
+    array.push(request)
+  },
+  error: error => console.error(error),
+  complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
+})
 
 
 

test/grpc1.v2.bak

@@ -1,168 +0,0 @@
-import { Subject, take } from 'rxjs';
-import { GrpcServiceMethod } from '../services/grpc.service.method';
-import { readFileSync } from 'fs';
-import { ConnectionRequest, Message } from '../interfaces/general.interface';
-import { ServerClientManager } from '../services/server-client.service';
-
-// Subject for bidirectional communication
-const connectionService: ServerClientManager = new ServerClientManager(new GrpcServiceMethod())
-const messagesJSON: any = readFileSync('payload.json')
-let parsedMessages: any[] = JSON.parse(messagesJSON) // load the fake messages generated for this trial 
-let targetserver: string = 'localhost:3001'
-let targetserver2: string = 'localhost:3002'
-let hostServer: string = 'localhost:3000'
-let array: any[] = [] // Used for testing                     
-let connectionRequest: ConnectionRequest = {
-  server: {
-    name: 'grpc1',
-    serverUrl: hostServer,
-    connectionType: 'GRPC',
-    messageToBePublishedfromApplication: new Subject<Message>()
-  },
-  client: [{
-    name: 'grpc2',
-    targetServer: targetserver,
-    connectionType: 'GRPC',
-    messageToBeReceivedFromRemote: new Subject<Message>()
-  }]
-}
-let client :ConnectionAttribute[] = [ 
-{ 
-  name:"con1"
-  ConnectionID: "aaa123-xxx123",
-  outGoing: {
-    Name?: string,
-    ChannelID?: "aaa123",
-    PublisherID?: "bbb123",
-    SubscriberID?: "ccc123", 
-}
-,
-  inComing: {
-    Name?: string,
-    ChannelID?: "xxx123",
-    PublisherID?: "yyy123",
-    SubscriberID?: "zzz123", 
-}
-,
-  connectionStatus: Subject<ReportStatus>
-}
-
-{ 
-  name:"con2"
-  ConnectionID: "aaa123xxx-xxx123xx",
-  outGoing: {
-    Name?: string,
-    ChannelID?: "aaa123xxx",
-    PublisherID?: "bbb123",
-    SubscriberID?: "ccc123xxx", 
-}
-,
-  inComing: {
-    Name?: string,
-    ChannelID?: "xxx123xx",
-    PublisherID?: "yyy123xxx",
-    SubscriberID?: "zzz123xxx", 
-}
-,
-  connectionStatus: Subject<ReportStatus>
-}
-]
-
-// Handler for the incoming Messages from the other side. 
-connectionRequest.client.forEach((client) => {
-  client.messageToBeReceivedFromRemote.subscribe({
-    next: request => {
-      // Application logic comes here. This is where the asortment takes place, of decidiing whose messages it belongs of what it is
-      if ((request.message as MessageLog).appData.msgPayload == 'Query') {
-        generateFakeStreamResponse(request).subscribe({
-          next: (responseMessage: Message) => {
-            // console.log(`Processing request:${request.id}....`)
-            connectionRequest.server.messageToBePublishedfromApplication.next(responseMessage)
-          },
-          error: error => console.error(error),
-          complete: () => {
-            console.log(`Stream request for ${request.id} is queued.`) // shpuld be indefinite
-          }
-        })
-      } else {
-        array.push(request)
-        console.log(`Received messages from the other side: ${(request.message as MessageLog).appData.msgId}`)
-      }
-    },
-    error: error => console.error(error),
-    complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
-  })
-})
-
-
-
-connectionService.generateConnection(connectionRequest)
-
-/* Simple Test */
-// let generateFakeMessagesToBePublished = stream().pipe(take(10))
-// generateFakeMessagesToBePublished.subscribe({
-//   next: message => {
-//     let payload: Message = {
-//       id: hostServer,
-//       message: message
-//     }
-//     connectionRequest.server.messageToBePublishedfromApplication.next(payload)
-//   }
-// })
-
-
-/* Complex Test: Expected out come, both must receive 14 message by the end. Havent try to disconnect.*/
-setTimeout(() => {
-  let message = {
-    id: parsedMessages[10].appData.msgId,
-    message: parsedMessages[10] // Choose this number, because i purposely use the 11th message and change the msgPayload property to query to emulate a request
-  }
-  connectionRequest.server.messageToBePublishedfromApplication.next(message)
-}, 3000)
-setTimeout(() => {
-  let message = {
-    id: parsedMessages[11].appData.msgId,
-    message: parsedMessages[11]// Choose this number, because i purposely use the 12th message and change the msgPayload property to query to emulate a request
-  }
-  connectionRequest.server.messageToBePublishedfromApplication.next(message)
-}, 4000)
-setTimeout(() => {
-  console.log(`All received data: ${array.length}`)
-}, 10000)
-setTimeout(() => {
-  console.log(`All received data: ${array.length}`)
-}, 20000)
-
-
-// this is just to publish an array of fake data as a Subject
-function stream(): Subject<any> {
-  let result: Subject<any> = new Subject()
-  let messages: any[] = parsedMessages
-  let count = 0
-  const intervalId = setInterval(() => {
-    result.next(messages[count]);
-    count++;
-    if (count >= 1000) {
-      clearInterval(intervalId);
-      result.complete();
-    }
-  }, 500)
-  return result
-}
-
-
-function generateFakeStreamResponse(request: any): Subject<any> {
-  let res: Subject<any> = new Subject()
-  stream().pipe(take(7)).subscribe({
-    next: element => {
-      let message = {
-        id: request.id, // Caller's 
-        message: element
-      }
-      res.next(message)
-    },
-    error: error => console.error(error),
-    complete: () => console.log(`Stream response for ${request.id} has been prepared.`)
-  })
-  return res
-}

test/grpc2.ts

@@ -8,7 +8,7 @@ import mongoose from 'mongoose';
 mongoose.connect('mongodb://localhost:27017/grpc2')
 const Message = mongoose.model('Message', require('../models/message.schema'))
 // Subject for bidirectional communication
-const connectionService: ServerClientManager = new ServerClientManager(new GrpcServiceMethod())
+const connectionService: ServerClientManager = new ServerClientManager()
 const messagesJSON: any = readFileSync('payload.json')
 let parsedMessages: any[] = JSON.parse(messagesJSON) // load the fake messages generated for this trial 
 let targetserver: string = 'localhost:3000'

test/grpc3.ts

@@ -1,24 +1,27 @@
-import { Subject, take } from 'rxjs';
+import { Subject, from, interval, take } from 'rxjs';
 import { Message, MessageLog, ConnectionRequest } from '../interfaces/general.interface';
 import { GrpcServiceMethod } from '../services/grpc.service.method';
 import { readFileSync } from 'fs';
 import { ServerClientManager } from '../services/server-client.service';
+import mongoose from 'mongoose';
 
+mongoose.connect('mongodb://localhost:27017/grpc2')
+const Message = mongoose.model('Message', require('../models/message.schema'))
 // Subject for bidirectional communication
-const connectionService: ServerClientManager = new ServerClientManager(new GrpcServiceMethod())
+const connectionService: ServerClientManager = new ServerClientManager()
 const messagesJSON: any = readFileSync('payload.json')
 let parsedMessages: any[] = JSON.parse(messagesJSON) // load the fake messages generated for this trial 
 let targetserver: string = 'localhost:3000'
 let targetserver2: string = 'localhost:3001'
 let hostServer: string = 'localhost:3002'
-let array: any[] = [] // Used for testing    
-let intervalToStreamOutGoingMessage: number = 10
+let intervalToStreamOutGoingMessage: number = 1
+let array: Message[] = []
 
 
 /* Simple Test: 1 to 1 */
 let connectionRequest: ConnectionRequest = {
   server: {
-    name: 'g2',
+    name: 'g3',
     serverUrl: hostServer,
     connectionType: 'GRPC',
     messageToBePublishedFromApplication: new Subject<Message>()
@@ -32,21 +35,29 @@ let connectionRequest: ConnectionRequest = {
 }
 
 connectionService.generateConnection(connectionRequest)
+// 10000th message == 848438e1-da50-4d98-aa12-e44d6d6a1489
+
+// let generateFakeMessagesToBePublished = stream().pipe(take(1000))
+// let generateFakeMessagesToBePublished = from(parsedMessages).pipe(take(1000))
+// generateFakeMessagesToBePublished.subscribe({
+//   next: message => {
+//     let payload: Message = {
+//       id: hostServer,
+//       message: message
+//     }
+//     connectionRequest.server.messageToBePublishedfromApplication.next(payload)
+//   }
+// })
 
-let generateFakeMessagesToBePublished = stream().pipe(take(10))
-generateFakeMessagesToBePublished.subscribe({
-  next: message => {
-    let payload: Message = {
-      id: hostServer,
-      message: message
-    }
-    connectionRequest.server.messageToBePublishedFromApplication.next(payload)
-  }
-})
 
 connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
   next: response => {
+    // if((response.message as MessageLog).appData.msgId == `ebf94479-44fe-470d-827c-9f1389396d6a`){
+    //   console.log(`Received the 1000th message. Running the test. Initiating server restart....`)
+    // connectionService.restartServerInDuration(10)
+    // }
     console.log(`Received ${(response.message as MessageLog).appData.msgId} from ${connectionRequest.client.targetServer}`)
+    // Message.create(response)
     array.push(response)
   },
   error: error => console.error(error),
@@ -54,7 +65,6 @@ connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
 })
 
 
-
 /* Complex Test: 1 to 1*/
 // let connectionRequest: ConnectionRequest = {
 //   server: {
@@ -323,13 +333,6 @@ function generateFakeStreamResponse(request: any): Subject<any> {
 }
 
 /* Checking the values by the end of the test */
-setTimeout(() => {
-  console.log(`All received data: ${array.length}`)
-}, 5000)
-setTimeout(() => {
-  console.log(`All received data: ${array.length}`)
-}, 10000)
-setTimeout(() => {
-  console.log(`All received data: ${array.length}`)
-}, 15000)
-
+interval(5000).subscribe(() => {
+  console.log(`All received data: ${array.length}`);
+});