retransmission service && new Utils && deprecate Server Client && more test cases

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "dependencies/logging"]
+	path = dependencies/logging
+	url = https://git.swopt.com/FAMBL/Fis-LoggingService.git

dependencies/logging

@@ -0,0 +1 @@
+Subproject commit da490bb47270703851bd44183175d0edbaa73a0c

interfaces/general.interface.ts

@@ -1,6 +1,7 @@
 /* General interface used for office work/ */
 
 import { Observable, Subject } from "rxjs"
+import { BaseMessage } from "../dependencies/logging/interface/export";
 
 export interface ConnectionState {
     uuid?: string | number;
@@ -78,3 +79,15 @@ export interface ConnectionID {
     remote: string
 }
 
+export interface ClientNotificationState {
+    event: string,
+    message: string,
+    status?: 'ONLINE' | 'OFFLINE' | null 
+}
+
+
+export interface WrappedMessage {
+    timeReceived: any, // this property is for sender to sort
+    payload: BaseMessage,
+    previousMessageID?: string // this property is for receiver to sort
+}

interfaces/retransmission.interface.ts

@@ -0,0 +1,5 @@
+import { Observable } from "rxjs";
+
+export interface RetransmissionInterface  {
+    retransmission(payloadToBeTransmitted: Observable<any>, eventListener: Observable<any>): Observable<any>
+}

package.json

@@ -9,12 +9,19 @@
     "test": "echo \"Error: no test specified\" && exit 1",
     "test2": "node test/test2.js",
     "generatedata": "node services/utility/generateData.js",
+    "http1": "node test/http1.js",
+    "http2": "node test/http2.js",
+    "socket": "node test/socket.js",
     "grpc1": "node test/grpc1.js",
     "grpc2": "node test/grpc2.js",
     "grpc3": "node test/grpc3.js",
     "testing": "node test/test.js",
     "server": "node test/grpcTest.js",
-    "simpleObsTest": "node test/simpleObsTest.js"
+    "rxjsbuffer": "node test/rxjsbuffer.sample.test.js",
+    "http-test-server": "node test/http-test-server.js",
+    "socket-test-server": "node test/socket-test-server.js",
+    "simpleObsTest": "node test/simpleObsTest.js",
+    "compareString": "node test/stringtest.js"
   },
   "author": "",
   "license": "ISC",
@@ -24,14 +31,21 @@
     "dotenv": "^16.3.1",
     "express": "^4.18.2",
     "grpc-health-check": "^2.0.0",
+    "jsonschema": "^1.4.1",
     "lodash": "^4.17.21",
     "mongo": "^0.1.0",
     "mongoose": "^7.6.0",
+    "rfdc": "^1.4.1",
     "rxjs": "^7.8.1",
+    "socket.io": "^4.7.5",
+    "socket.io-client": "^4.7.5",
     "uuid": "^9.0.1"
   },
   "devDependencies": {
+    "@types/jest": "^29.5.12",
     "@types/node": "^20.6.0",
-    "typescript": "^5.2.2"
+    "jest": "^29.7.0",
+    "ts-jest": "^29.2.5",
+    "typescript": "^5.5.4"
   }
 }

services/retransmission.service.ts

@@ -0,0 +1,128 @@
+import { BehaviorSubject, buffer, concatMap, distinctUntilChanged, from, Observable, Subject, takeWhile } from "rxjs";
+import { BaseMessage } from "../dependencies/logging/dependencies/msgutil/dependencies/dependencies";
+import { RetransmissionInterface } from "../interfaces/retransmission.interface";
+import { WrappedMessage } from "../interfaces/general.interface";
+import { sortMessageBasedOnDate } from "./utility/message-ordering";
+
+export class RetransmissionService implements RetransmissionInterface {
+    private sortMessage: boolean = false
+    private bufferReleaseSignal: Subject<void> = new Subject()
+    private receiverConnectionState: BehaviorSubject<'OFFLINE' | 'ONLINE'> = new BehaviorSubject('OFFLINE')
+    private transmissionState: BehaviorSubject<'TRANSMITTING' | 'IDLE' | 'ARRAY EMPTY' | 'STORING DATA' | 'GETTING STORED DATA'> = new BehaviorSubject('ARRAY EMPTY')
+    private arrayToBeTransmitted: Subject<WrappedMessage[]> = new Subject()
+    private toBeWrapped: Subject<any> = new Subject()
+    private wrappedMessageToBeBuffered: Subject<WrappedMessage> = new Subject()
+    private messageToBeTransmitted: Subject<WrappedMessage> = new Subject()
+
+    // Interface
+    public retransmission(payloadToBeTransmitted: Observable<any>, eventListener: Observable<any>, messageOrdering?: boolean): Observable<any> {
+        return new Observable((observer) => {
+            if (messageOrdering) {
+                this.sortMessage = true
+                console.log(`Message ordering is set to ${this.sortMessage}`)
+            }
+            eventListener.subscribe(event => this.receiverConnectionState.next(event))
+
+            this.startWrappingOperation()
+            this.startBufferTransmisionProcess()
+            this.releaseSignalManager()
+
+            payloadToBeTransmitted.subscribe((message) => {
+                this.toBeWrapped.next(message)
+            })
+
+            this.messageToBeTransmitted.subscribe(message => observer.next(message))
+        })
+    }
+
+    private startWrappingOperation() {
+        let currentMessageId: string | null
+        this.toBeWrapped.subscribe(message => {
+            this.wrappedMessageToBeBuffered.next(this.wrapMessageWithTimeReceived(message, currentMessageId ? currentMessageId : null))
+            currentMessageId = message.header.messageID
+        })
+        //simulate connection test
+
+        // wrappedMessageToBeBuffered will then be pushed to buffer
+        this.wrappedMessageToBeBuffered.pipe(buffer(this.bufferReleaseSignal)).subscribe((bufferedMessages: WrappedMessage[]) => {
+            console.log(bufferedMessages.length + ' buffered messages')
+            console.log(`Released buffered message: ${bufferedMessages.length} total messages. To Be sorted.`)
+            // arrayToBeTransmitted.next(sortMessage(bufferedMessages))
+            this.arrayToBeTransmitted.next(this.sortMessage && bufferedMessages.length > 0 ? sortMessageBasedOnDate(bufferedMessages) : [])
+        });
+    }
+
+    private wrapMessageWithTimeReceived(message: any, previousMessageID: string): WrappedMessage {
+        // check if message has already a time received property if so no need to add anymore
+        if (!message.timeReceived) {
+            let WrappedMessage: WrappedMessage = {
+                timeReceived: new Date(),
+                payload: message as BaseMessage,
+                previousMessageID: previousMessageID
+            }
+            return WrappedMessage
+        } else {
+            return message as WrappedMessage
+        }
+    }
+
+    private startBufferTransmisionProcess() {
+        console.log(`StartBufferTransmissionProcess`)
+        this.arrayToBeTransmitted.subscribe(array => {
+            if (array.length > 0) {
+                this.transmissionState.next('TRANSMITTING')
+                from(array).subscribe({
+                    next: (message: WrappedMessage) => {
+                        if (this.receiverConnectionState.getValue() == 'OFFLINE') {
+                            // buffer this message. Flush it back to buffer
+                            this.wrappedMessageToBeBuffered.next(message)
+                        }
+                        if (this.receiverConnectionState.getValue() == 'ONLINE') {
+                            this.messageToBeTransmitted.next(message)
+                        }
+                    },
+                    error: err => console.error(err),
+                    complete: () => {
+                        // update transmission state to indicate this batch is completed
+                        console.log(`Processing buffered array completed. Changing transmission state to ARRAY EMPTY`);
+                        this.transmissionState.next('ARRAY EMPTY');
+
+                        if (this.receiverConnectionState.getValue() === 'ONLINE' && this.transmissionState.getValue() === 'ARRAY EMPTY') {
+                            setTimeout(() => {
+                                this.bufferReleaseSignal.next()
+                            }, 1000)
+                        }
+                        // Do nothing if the receiver connection is offline
+                    }
+                });
+            } else {
+                // If I don't do setTimeout, then bufferrelasesignal will be overloaded
+                setTimeout(() => {
+                    this.bufferReleaseSignal.next()
+                }, 3000)
+            }
+        }
+        )
+    }
+
+    private releaseSignalManager() {
+        this.receiverConnectionState.pipe(
+            distinctUntilChanged()
+        ).subscribe(clientState => {
+            console.log(`Client is now ${clientState}`)
+            if (clientState == 'OFFLINE') {
+                console.log(`Current transmission state: ${this.transmissionState.getValue()}`)
+                // just keep buffering
+            }
+            if (clientState == 'ONLINE') {
+                console.log(`Current transmission state: ${this.transmissionState.getValue()}`)
+                // get the stored messages to pump it back into the buffer to be ready to be processed immediately
+                if (this.transmissionState.getValue() == 'ARRAY EMPTY') {
+                    this.bufferReleaseSignal.next()
+                }
+
+            }
+        })
+    }
+}
+

services/server-client.service.ts

@@ -6,6 +6,10 @@ import * as dotenv from 'dotenv'
 import * as _ from 'lodash'
 dotenv.config()
 
+/**
+ * @deprecated This service will be removed in future versions. 
+ * We are using websocket now.
+ */
 export class ServerClientManager {
     private connectionAttributes: ConnectionAttribute[] = []
     private grpcService: GrpcServiceMethod = new GrpcServiceMethod()

services/storage.service.ts

@@ -0,0 +1,79 @@
+import { from, Observable, Subject, Subscriber, Subscription } from "rxjs";
+import { BaseMessage, LoggingService } from "../dependencies/logging/services/logging-service";
+import { DateStructure, LogSetting, MsgDateTime } from "../dependencies/logging/type/datatype";
+import { MessageLog, ClientNotificationState } from "../interfaces/general.interface";
+
+// From Chin: "If you can add feature to set buffering by disconnect duration and buffer size that would be good.""
+export class StorageService {
+    private msgForLogging: Subject<MessageLog | BaseMessage> = new Subject()
+    private loggingSubscription: Subscription | null = null
+    private loggingService: LoggingService 
+    //var for logging filter
+    private bufferStart!: Date
+    private bufferEnd!: Date
+
+    constructor(logSettings: LogSetting) {
+        // need to first initialize logging service
+        this.loggingService = new LoggingService()
+        this.loggingService.init(logSettings)
+        this.loggingService.subscribe(this.msgForLogging).then(() => {
+        }).catch((error) => console.error(error))
+    }
+
+    /* Will first subscribe to message stream, and then start buffering all the data into the designated buffer. */
+    public async disconnectionHandler(bufferedMessage: BaseMessage[]): Promise<any> {
+        return new Promise((resolve, reject) => {
+            let messageCount: number = 0
+            this.bufferStart = new Date()
+            if (!this.loggingSubscription || this.loggingSubscription.closed) {
+                this.loggingSubscription = from(bufferedMessage).subscribe(message => {
+                    this.msgForLogging.next(message)                
+                })
+                resolve({ event: 'Source Subscrpition', message: 'Logging standing by. can release from rxjs buffer now', sourceSubscribed: true })
+            } else {
+                console.log(`Retransmission: Has already assigned logging Subscription`)
+                // if things goes well, am not supposed to see this line
+            }
+        })
+    }
+
+    public reconnectionHandler(): Observable<BaseMessage> {
+        return new Observable((bufferedMessages: Subscriber<BaseMessage>) => {
+            this.bufferEnd = new Date()
+            // destroy previous subscription
+            if (this.loggingSubscription) {
+                this.loggingSubscription.unsubscribe()
+                // console.log(this.loggingSubscription)
+
+                // for more precise control, since there's no delete log in logging service, and we do not want to extract previously released buffered messages
+                let dateRange: MsgDateTime = {
+                    from: this.createDateStructure(this.bufferStart),
+                    to: this.createDateStructure(this.bufferEnd)
+                }
+
+                this.loggingService.filter(dateRange).then((array) => {
+                    console.log(array.length)
+                    array.forEach(message => {
+                        bufferedMessages.next(JSON.parse(message.appData.msgPayload as string) as BaseMessage)
+                    })
+                    bufferedMessages.complete()
+                }).catch((error) => {
+                    console.error(error)
+                    bufferedMessages.error('Something went wrong')
+                })
+            } else {
+                bufferedMessages.complete()
+            }
+        })
+    }
+
+    private createDateStructure(date: Date): DateStructure {
+        return {
+            date: date.toISOString().split('T')[0], // Store date as a string in YYYY-MM-DD format
+            hour: date.getHours(),
+            minute: date.getMinutes(),
+            second: date.getSeconds()
+        };
+    }
+
+}

services/utility/buffer-obs.ts

@@ -0,0 +1,198 @@
+import { buffer, bufferCount, bufferTime, bufferToggle, catchError, combineLatest, filter, finalize, from, map, mapTo, merge, mergeAll, Observable, of, startWith, Subject, Subscriber, Subscription, switchMap, take, takeUntil, tap, throwError, timer } from "rxjs";
+import { ClientNotificationState } from "../../interfaces/general.interface";
+import { BaseMessage } from "../../dependencies/logging/services/logging-service";
+
+export function rxjsBufferToggle(sourceObs: Subject<BaseMessage>, notificationObs: Subject<ClientNotificationState>): Observable<BaseMessage | BaseMessage[]> {
+    let releaseTrigger = new Subject<any>()
+    let alreadyDisconnected = false
+    const messageStream: Observable<BaseMessage | BaseMessage[]> = notificationObs.pipe(
+        map(notif => {
+            if (notif.message == 'Disconnected' && !alreadyDisconnected) {
+                alreadyDisconnected = true
+                return true
+            }
+            // if (notif.message == 'Subscribed' || notif.status == 'ONLINE' || notif.message == 'Reconnected') {
+            else if (notif.message == 'Subscribed') {
+                return false
+            } else {
+                return null
+            }
+        }),
+        switchMap(toBuffer => {
+            if (toBuffer) {
+                console.log(`Rxjs Buffering...`)
+                return sourceObs.pipe(
+                    // tap(value => console.log(`rxjs buffering ${value?.header?.messageID ?? 'unknown'}`)),
+                    bufferToggle(
+                        // bufferTrigger.pipe(startWith(0)),
+                        // of(0),
+                        from([0]),
+                        () => releaseTrigger
+                    ));
+            } else {
+                console.log(`Rxjs Releasing...`)
+                return sourceObs
+            }
+        })
+    )
+    // only for triggering when to stop buffering
+    notificationObs.subscribe((notification: ClientNotificationState) => {
+        let status: 'Online' | 'Offline-NotSubscribed' | 'Offline-Subscribed' | null = null
+        if (notification.message == 'Reconnected') {
+            status = 'Online'
+        }
+        if (notification.message == 'Disconnected' && status != `Offline-NotSubscribed`) {
+            status = 'Offline-NotSubscribed'
+        }
+        // Technically speaking, we only ever need this, just for the release signal. 
+        if (notification.message == 'Subscribed' && status != 'Offline-Subscribed') {
+            releaseTrigger.next('Release')
+            status = 'Offline-Subscribed'
+        }
+        if (notification.message == 'Subscribed' && status != 'Offline-NotSubscribed') {
+            console.log(`Retransmission already subscribed to buffer output. `)
+        }
+    })
+    return messageStream
+}
+
+
+// https://rxjs.dev/api/index/function/buffer
+export function rxjsBufferDefault(sourceObs: Subject<BaseMessage>, releaseSignal: Observable<any>): Observable<BaseMessage[]> {
+    return new Observable((observer: Subscriber<BaseMessage[]>) => {
+        let bufferedMessages = sourceObs.pipe(buffer(releaseSignal));
+        bufferedMessages.subscribe((bufferedMessages: BaseMessage[]) => {
+            observer.next(bufferedMessages)
+        });
+    })
+}
+
+// https://rxjs.dev/api/index/function/bufferTime
+export function rxjsBufferTime(sourceObs: Subject<BaseMessage>, duration: number): Observable<BaseMessage[]> {
+    return new Observable((observer: Subscriber<BaseMessage[]>) => {
+        let bufferedMessages = sourceObs.pipe(bufferTime(duration))
+        bufferedMessages.subscribe((bufferedMessages: BaseMessage[]) => {
+            observer.next(bufferedMessages)
+        })
+    })
+}
+
+// https://rxjs.dev/api/index/function/bufferCount
+export function rxjsBufferCount(sourceObs: Subject<BaseMessage>, bufferlimit: number): Observable<BaseMessage[]> {
+    return new Observable((observer: Subscriber<BaseMessage[]>) => {
+        let bufferedMessages = sourceObs.pipe(bufferCount(bufferlimit))
+        bufferedMessages.subscribe((bufferedMessages: BaseMessage[]) => {
+            observer.next(bufferedMessages)
+        })
+    })
+}
+
+// Only the first one is working. The others like buffer duration and buffersize limit is still faulty. Please refrain from using these for now.
+export function rxjsBuffer(sourceObs: Subject<BaseMessage>, releaseSignal: Observable<any>, bufferDuration?: number, bufferSizeLimit?: number, messageToBeLogged?: Subject<BaseMessage>): Observable<BaseMessage[]> {
+    return new Observable((observer: Subscriber<BaseMessage[]>) => {
+        if (!bufferDuration && !bufferSizeLimit) {
+            console.log(`Default Rxjs Buffering....`)
+            sourceObs.pipe(buffer(releaseSignal)).subscribe((bufferedMessages: BaseMessage[]) => {
+                observer.next(bufferedMessages)
+            });
+        }
+        if (bufferDuration && !bufferSizeLimit) {
+            console.log(`Rxjs Buffering with bufferDuration: ${bufferDuration}`)
+            // Observable that emits after a certain period (e.g., 5 seconds)
+            let stopBuffering = timer(bufferDuration)
+            // let stopBuffering = () => timer(bufferDuration)
+            let releaseBuffering = merge(stopBuffering, releaseSignal)
+            sourceObs.pipe(
+                bufferToggle(of(0), () => releaseBuffering)
+            ).subscribe(((bufferedMessage: BaseMessage[]) => {
+                console.log(`Buffering Duration Exceeded. Please don't send me any more data.`)
+                // Should emit an event that duration limit has been exceeded
+                if (messageToBeLogged) {
+                    bufferedMessage.forEach((message: BaseMessage) => {
+                        messageToBeLogged.next(message)
+                    })
+                } else {
+                    observer.next(bufferedMessage)
+                }
+            }))
+        }
+        if (!bufferDuration && bufferSizeLimit) {
+            console.log(`rxjs Buffering with buffer limit: ${bufferSizeLimit}`)
+            sourceObs.pipe(
+                bufferCount(bufferSizeLimit),
+                take(1), // take(1) then takes that first buffered array and completes the observable, meaning it will not allow any further emissions or buffering.
+                tap(bufferedMessage => {
+                    if (bufferedMessage.length > bufferSizeLimit) {
+                        throw new Error('Amount exceeded')
+                    } else {
+                        // keep buffering
+                    }
+                }),
+                catchError(err => {
+                    console.error(err)
+                    // maybe emit an event for this as well
+                    return throwError(() => new Error('Rxjs Buffering STOP!'))
+                })
+            ).subscribe({
+                next: (bufferedMessage: BaseMessage[]) => {
+                    if (messageToBeLogged) {
+                        from(bufferedMessage).subscribe({
+                            next: (message: BaseMessage) => {
+                                messageToBeLogged.next(message)
+                            },
+                            error: err => console.error(err),
+                            complete: () => {
+                                // emit event to indicate storage logging is completed
+                            }
+                        })
+                    }
+                    observer.next(bufferedMessage)
+                },
+                error: err => {
+                    // console.error(err)
+                    observer.error(err)
+                },
+                complete: () => {
+                    observer.complete()
+                }
+            })
+        }
+
+        if (bufferDuration && bufferSizeLimit) {
+            console.log(`rxjs Buffering with bufferDuration: ${bufferDuration} ms && bufferSizeLimit: ${bufferSizeLimit} items`);
+
+            let islimitExceeded = false
+            let bufferedArray: BaseMessage[] = []
+            // buffer duration limit.
+            timer(bufferDuration).subscribe(() => islimitExceeded = true)
+            // buffer size limit. 
+            sourceObs.pipe(
+                bufferCount(bufferSizeLimit),
+                take(1) // this take will close the subscription as well, preventing memory leak
+            ).subscribe(() => islimitExceeded = true)
+
+            let subscription: Subscription = sourceObs.subscribe(message => {
+                if (!islimitExceeded) {
+                    bufferedArray.push(message)
+                } else {
+                    if (messageToBeLogged) {
+                        from(bufferedArray).subscribe({
+                            next: (message: BaseMessage) => {
+                                messageToBeLogged.next(message)
+                            },
+                            error: err => console.error(err),
+                            complete: () => {
+                                subscription.unsubscribe()
+                                // emit event to indicate storage logging is completed
+                            }
+                        })
+                    } else {
+                        console.log(`bufferArray exceeded. Removing all buffered messages`)
+                        // observer.next(bufferedArray)
+                        subscription.unsubscribe()
+                    }
+                }
+            })
+        }
+    })
+}

services/utility/message-ordering.ts

@@ -0,0 +1,8 @@
+import { WrappedMessage } from "../../interfaces/general.interface";
+
+export function sortMessageBasedOnDate(array: WrappedMessage[]): WrappedMessage[] {
+    console.log(`Sorting ${array.length} messages....`)
+    return array.sort((a, b) => {
+        return new Date(a.timeReceived).getTime() - new Date(b.timeReceived).getTime();
+    });
+}

services/utility/prepareFISmessage.ts

@@ -0,0 +1,48 @@
+import { interval, Observable, Subject, Subscription } from "rxjs"
+import { FisCreateMessageUtility } from "../../dependencies/logging/dependencies/msgutil/services/fis.create.message"
+import { BaseMessage } from "../../dependencies/logging/services/logging-service"
+
+let messageUtil: FisCreateMessageUtility = new FisCreateMessageUtility("FisAppID/Name")
+export function prepareResponseMessages(amount: number, intervalBetweenMessages?: number): Subject<BaseMessage> {
+
+    let responseData: any[] = []
+    let returnSubject = new Subject<BaseMessage>()
+    for (let i = 0; i < 1000; i++) {
+        responseData.push(`bla bla bla`)
+    }
+    let count = 0
+    let subscription: Subscription
+
+    if (intervalBetweenMessages) {
+        console.log(`Generate data with intervals`)
+        subscription = interval(intervalBetweenMessages).subscribe({ // 10millisecond
+            next: () => {
+                count++
+                if (count <= amount) {
+                    let message = messageUtil.getResponseDataMessage('123', responseData, messageUtil.getLoginMessage())
+
+                    returnSubject.next(message)
+                } else {
+                    returnSubject.complete()
+                    subscription.unsubscribe()
+                }
+            }
+        });
+    } else {
+        console.log(`Generate data without intervals`)
+        for (let i = 0; i <= amount; i++) {
+            if (i >= amount) {
+                returnSubject.complete()
+            } else {
+                let message = messageUtil.getResponseDataMessage('123', responseData, messageUtil.getLoginMessage())
+                returnSubject.next(message)
+            }
+        }
+    }
+    return returnSubject
+}
+
+
+export function generateSingleMessage() {
+    return messageUtil.getResponseDataMessage('123', [], messageUtil.getLoginMessage())
+}

starthttp.bat

@@ -0,0 +1,7 @@
+
+@echo off
+start wt -M -d "E:\Task\Fis-Restransmission" cmd /k "npm run http-test-server" ; split-pane -d "E:\Task\Fis-Restransmission" cmd /k "npm run http2"
+
+
+//wt -p "Command Prompt" ; split-pane -p "Windows PowerShell" ; split-pane -H 
+

test/http-test-server.ts

@@ -0,0 +1,53 @@
+// import express from 'express';
+const express = require('express')
+import { Request, Response } from 'express';
+import { BaseMessage, LoggingService } from '../dependencies/logging/services/logging-service';
+import { LogSetting } from '../dependencies/logging/type/datatype';
+import { Subject } from 'rxjs';
+
+const app = express();
+const port = 9999;
+let count = 0
+let logSettings: LogSetting = {
+  cacheMessageLimit: 0,
+  storage: "MongoDB",
+  setting: {
+    appName: 'Retransmission',
+    appLocName: 'Local Mongo Buffer',
+    logLocName: 'Local Mongo BUffer',
+  },
+  customSetting: {
+    server: "localhost:27017",
+    database: 'client'
+  }
+}
+let logginService = new LoggingService()
+let incomingMessage: Subject<BaseMessage> = new Subject()
+
+logginService.init(logSettings).then(() => {
+  logginService.subscribe(incomingMessage)
+}).catch((err) => console.error(err))
+
+
+// Middleware to parse JSON bodies
+app.use(express.json());
+
+// Handling a GET request
+app.get('/', (req: Request, res: Response) => {
+  res.send('Hello, I am alive!');
+});
+
+// Handling a POST request
+app.post('/data', (req: Request, res: Response) => {
+  const data = req.body;
+  count++
+  console.log(`${count} : ${data?.header?.messageID ?? 'undefined Data'}`)
+  incomingMessage.next(data)
+  // Send a JSON response
+  res.json({ message: `Received data`, data: data });
+  // res.send(`Received data: ${data?.header?.messageID ?? 'undefined Data'}`);
+});
+
+app.listen(port, () => {
+  console.log(`Server is running at http://localhost:${port}`);
+});

test/http1.ts

@@ -0,0 +1,112 @@
+import { interval as RxjsInterval, Subject, Subscription } from "rxjs"
+import { BehaviorSubject } from "rxjs"
+import { BufferService } from "../services/buffer.service"
+import { ConnectionState, Message } from "../interfaces/general.interface"
+import { v4 as uuidv4 } from 'uuid'
+import { error } from "console"
+
+console.log(`Testing for HTTP buffer service.`)
+
+let source: Subject<Message> = new Subject()
+let initialReport: ConnectionState = { status: 'DIRECT_PUBLISH' }
+let connectionStateSubject: BehaviorSubject<ConnectionState> = new BehaviorSubject(initialReport)
+let bufferService: BufferService = new BufferService(source, connectionStateSubject, 'test')
+
+/* So, have an interval obseravable that will post a method every second, but it will be buffered instead of being post.
+or something like that. */
+
+// Create an Observable that emits something every 1 second
+const interval = RxjsInterval(1000);
+interval.subscribe({
+    next: time => {
+        let message = {
+            id: uuidv4(),
+            message: `I am to be posted`
+        }
+        source.next(message)
+    }
+})
+
+bufferService.getMessages().subscribe({
+    next: message => {
+        // Usage example:
+        fetch('http://localhost:9999/data', {
+            method: 'POST',
+            body: JSON.stringify(message),
+            headers: {
+                "Content-type": "application/json; charset=UTF-8"
+            }
+        }).then((response) => {
+            console.log(`sending ${message.id}`)
+            console.log(response.status)
+            connectionStateSubject.next({ status: 'DIRECT_PUBLISH' })
+        }).catch((error) => {
+            console.error(error)
+            connectionStateSubject.next({ status: 'BUFFER' })
+            periodicCheck()
+        })
+
+    }
+})
+
+
+
+function periodicCheck() {
+    let timer = RxjsInterval(1000).subscribe({
+        next: everySecond => {
+            fetch('http://localhost:9999/', {
+                method: 'GET',
+                headers: {
+                    "Content-type": "application/json; charset=UTF-8"
+                }
+            }).then((response) => {
+                if (response.ok) {
+                    connectionStateSubject.next({ status: 'DIRECT_PUBLISH' })
+                    timer.unsubscribe()
+                } else {
+                    connectionStateSubject.next({ status: 'BUFFER' })
+                }
+            }).catch((error) => {
+                connectionStateSubject.next({ status: 'BUFFER' })
+            })
+
+        }
+    })
+}
+// async function postData(url, data): Promise<any> {
+//     return new Promise(async (resolve, reject) => {
+//         try {
+//             const response = await fetch(url, {
+//                 method: 'POST',
+//                 mode: 'cors',
+//                 cache: 'no-cache',
+//                 credentials: 'same-origin',
+//                 headers: {
+//                     'Content-Type': 'application/json'
+//                 },
+//                 redirect: 'follow',
+//                 referrerPolicy: 'no-referrer',
+//                 body: JSON.stringify(data)
+//             });
+
+//             if (!response.ok) {
+//                 // throw new Error(`HTTP error! Status: ${response.status}`);
+//                 reject(`HTTP error! Status: ${response.status}`);
+//             }
+
+//             let responseData;
+//             try {
+//                 responseData = await response.json();
+//             } catch (jsonError) {
+//                 // throw new Error('Failed to parse JSON response');
+//                 reject('Failed to parse JSON response');
+//             }
+
+//             resolve(responseData);
+//         } catch (error) {
+//             console.error('Error making POST request:', error);
+//             throw error;
+//         }
+//     })
+// }
+

test/http2.ts

@@ -0,0 +1,65 @@
+/* TEST CASE FOR HTTP with Retransmission integrating event driven development */
+
+import { BehaviorSubject, buffer, filter, interval, map, Observable, startWith, Subject, Subscriber, Subscription, switchMap, takeUntil, tap } from "rxjs";
+import { ClientNotificationState } from "../interfaces/general.interface";
+import { rxjsBuffer } from "../services/utility/buffer-obs";
+import { LogSetting } from "../dependencies/logging/type/datatype";
+import { BaseMessage } from "../dependencies/logging/services/logging-service";
+import { prepareResponseMessages } from "../services/utility/prepareFISmessage";
+import { StorageService } from "../services/storage.service";
+import { RetransmissionService } from "../services/retransmission.service";
+
+
+let initialState: ClientNotificationState = { event: 'Initialization', message: 'InitialState', status: 'ONLINE' }
+let clientState: 'ONLINE' | 'OFFLINE' | null = 'ONLINE' // First assumption
+let logSettings: LogSetting = {
+    cacheMessageLimit: 0,
+    storage: "MongoDB",
+    setting: {
+        appName: 'Retransmission',
+        appLocName: 'Local Mongo Buffer',
+        logLocName: 'Local Mongo BUffer',
+    },
+    customSetting: {
+        server: "localhost:27017",
+        database: 'sender'
+    }
+}
+let sender: Subject<BaseMessage> = prepareResponseMessages(3000, 1) // generate fake messages second argument is milliseconds interval
+let finalOutput: Subject<BaseMessage> = new Subject()
+
+let notificationSubject: BehaviorSubject<ClientNotificationState> = new BehaviorSubject(initialState)
+let retransmissionService: RetransmissionService = new RetransmissionService()
+
+setTimeout(() => {
+    console.log(`Emulate Offline NOW`)
+    notificationSubject.next({
+        event: 'Test',
+        message: 'test',
+        status: 'OFFLINE'
+    })
+}, 3000);
+setTimeout(() => {
+    console.log(`Emulate Online NOW`)
+    notificationSubject.next({
+        event: 'Test',
+        message: 'test',
+        status: 'ONLINE'
+    })
+}, 6000);
+
+// retransmissionService.buffer(sender, notificationSubject, 1000, null, null).subscribe({
+//     next: (message: BaseMessage) => {
+//         finalOutput.next(message)
+//     },
+//     error: err => console.error(err),
+//     complete: () => { }
+// })
+
+let finalCount = 0
+finalOutput.subscribe(message => {
+    finalCount++
+    console.log(`final Output Count: ${finalCount}`)
+    // console.log(message.header.messageID)
+})
+

test/rxjsbuffer.sample.ts

@@ -0,0 +1,173 @@
+import { fa } from "@faker-js/faker"
+import { Subject, interval, Observable, buffer, switchMap, bufferToggle, map, scan, startWith, BehaviorSubject, tap } from "rxjs"
+import { prepareResponseMessages } from "../services/utility/prepareFISmessage"
+
+let toggle = interval(5000)
+const syncSource = interval(500) // sync
+let asyncSource = new Subject<any>() //async
+syncSource.subscribe(e => {
+    asyncSource.next(e)
+})
+
+/* So, output is the final and syncSource is the input. So the first trigger should triggle buffer.
+so initiailly, source should push to output, when toggle is triggered, source should be unsubscribed,
+and buffered should be instantiated immediately to keep receiving the messages. 
+When the next trigger comes in (reconnection), than connect source to output again, whilst aslo
+releasing buffered into output as well. */
+
+// assuming the initial state is online, so source will stream straight to output, but the first toggle will be offline
+/* How buffer works: It doesn't have a normal. It will subscribe to another observable that will act as a signal to release
+It's default state would be that it will always buffer, until the subscribed observable emits a signal to relase the buffer,
+and then it will continue to buffer. There's no way to adjust buffer to stream normally. */
+function bufferTest1() {
+    const intervalEvents = interval(1000);
+    const buffered = intervalEvents.pipe(buffer(toggle));
+    buffered.subscribe(x => console.log(x));
+}
+
+// VERSION 2 <with the help of chatGPT>
+// Toggle between buffering and normal emitting
+
+
+function bufferTest2() {
+    const toggleBuffer$ = toggle.pipe(
+        // Track the toggle state: true means buffering, false means normal
+        scan((isBuffering) => !isBuffering, false),
+
+        // Use the state to switch between buffer mode and normal mode
+        switchMap(isBuffering => {
+            // check for notif messatge and mutate open and close and isbuffering
+            // open.next(blablabla) or close.next(blablabla)
+            if (isBuffering) {
+                console.log(isBuffering)
+                // Start buffering: open on toggle, close on next toggle
+                return asyncSource.pipe(
+                    // bufferToggle(toggle, () => toggle)
+                    bufferToggle(toggle.pipe(startWith(0)), () => toggle))
+                // bufferToggle(open, () => { if (true) { return closing } })
+            } else {
+                console.log(isBuffering)
+                // Emit values normally
+                return asyncSource;
+            }
+        })
+    );
+
+    // Subscribe to the toggled stream
+    toggleBuffer$.subscribe(values => {
+        console.log('Values:', values);
+    });
+}
+
+// bufferTest2()
+// bufferTest1()
+
+let notificationSubject = new BehaviorSubject<string>('online') // true as in online false as in offline
+let keep = new Subject<any>()
+let release = new Subject<any>()
+// bufferTest3().subscribe(values => {
+//     console.log(`Values: ${values}`)
+// })
+
+// emulateNotification(`offline`, 3000)
+// emulateNotification(`online`, 7000)
+// emulateNotification(`offline`, 10000)
+// emulateNotification(`online`, 15000)
+
+function bufferTest3(): Observable<any> {
+    const messageStream = notificationSubject.pipe(
+        map(notif => {
+            if (notif == 'offline') {
+                return true
+            } else {
+                return false
+            }
+        }),
+        switchMap(notif => {
+            if (notif) {
+                // Start buffering: open on toggle, close on next toggle
+                return asyncSource.pipe(
+                    bufferToggle(keep.pipe(startWith(0)), () => release))
+            } else {
+                // Emit values normally
+                return asyncSource;
+            }
+        })
+    )
+    notificationSubject.subscribe(notif => {
+        // logic here
+        if (notif == 'online') {
+            console.log(`received notification: ${notif}, releasing...`)
+            release.next('release')
+        }
+
+        if (notif == 'offline') {
+            console.log(`received notification: ${notif}, buffering...`)
+            keep.next('keep')
+        }
+    })
+    return messageStream
+}
+
+function emulateNotification(status: string, delay: number) {
+    setTimeout(() => {
+        notificationSubject.next(status)
+    }, delay)
+}
+
+// working version
+function bufferTest4(): Observable<any> {
+    // Track buffering state
+    const bufferingState$ = notificationSubject.pipe(
+        scan((isBuffering, notif) => notif === 'offline' ? true : false, false),
+        tap(isBuffering => {
+            console.log(`Buffering state changed: ${isBuffering}`);
+        })
+    );
+
+    // Message stream based on buffering state
+    const messageStream = bufferingState$.pipe(
+        switchMap(isBuffering => {
+            if (isBuffering) {
+                // Start buffering: open on toggle, close on next toggle
+                return asyncSource.pipe(
+                    // bufferToggle(toggle.pipe(startWith(0)), () => release)
+                    bufferToggle(keep.pipe(startWith(0)), () => release)
+                );
+            } else {
+                // Emit values normally
+                return asyncSource;
+            }
+        })
+    );
+
+    notificationSubject.subscribe(notif => {
+        console.log(`Received notification: ${notif}`);
+        if (notif === 'online') {
+            release.next(true);  // Release buffered values
+        }
+
+        if (notif === 'offline') {
+            keep.next(true);  // Start buffering
+        }
+    });
+
+    return messageStream;
+}
+
+
+let messages = prepareResponseMessages(1000000, 1)
+function bufferTest5() {
+    const clicks = interval(300000);
+    const buffered = messages.pipe(buffer(clicks));
+    let count = 0
+    buffered.subscribe({
+        next: x => {
+            console.log(`${count++} && Buffer Length: ${x.length}`)
+        },
+        error: err => console.error(err),
+        complete: () => { }
+    })
+}
+
+bufferTest5()

test/socket-test-server.ts

@@ -0,0 +1,120 @@
+import { BehaviorSubject, from, Observable, Subject, takeUntil, takeWhile } from "rxjs";
+import { RetransmissionService } from "../services/retransmission.service";
+import { prepareResponseMessages } from "../services/utility/prepareFISmessage";
+import { BaseMessage } from "../dependencies/logging/services/logging-service";
+
+const express = require('express');
+const http = require('http');
+const { Server } = require('socket.io');
+
+const app = express();
+const server = http.createServer(app);
+const io = new Server(server);
+// Keep track of connected clients
+const clients: any[] = []
+let announcements: Subject<any> = new Subject()
+announcements.subscribe(announcement => {
+    console.log(`Server Announcement: ${announcement}`)
+})
+
+app.use(express.static('public'));
+
+io.on('connection', (socket) => {
+    announcements.next('a client connected:' + socket.id);
+    let clientInfo: ClientInfo = {
+        id: socket.id,
+        connectedAt: new Date(),
+        clientConnectionState: new BehaviorSubject<'ONLINE' | 'OFFLINE'>('ONLINE'),
+        requests: [],
+        buffer: new RetransmissionService()
+    }
+    let serverBuffer = new Subject<any>()
+    clients.push(clientInfo);
+    clientInfo.buffer.retransmission(serverBuffer, clientInfo.clientConnectionState).subscribe(output => socket.emit('message', output))
+
+    // Listen for messages from the client
+    socket.on('message', (request) => {
+        announcements.next(`Received Message: ${request.header.messageID} from ${clientInfo.id}`);
+        clientInfo.requests.push({ message: request, completed: false })
+        returnResponse(request).subscribe({
+            next: message => serverBuffer.next(message),
+            error: err => console.error(err),
+            complete: () => {
+                let clientOBJ = clientInfo.requests.find(obj => obj.message.header.messageID === request.header.messageID)
+                clientOBJ.completed = true
+                console.log('Current Array', clients)
+            }
+        })
+    });
+
+    socket.on('connect', (msg) => {
+        // Send a response back to the client
+        socket.emit('notification', `Hello from server. You have been assigned ${socket.id}`);
+    });
+
+    socket.on('interval', (value) => {
+        console.log(socket.id, value) // okay so it does receive in sequence after reconnection
+    })
+
+    // Handle disconnection
+    socket.on('disconnect', () => {
+        announcements.next(`Client ${clientInfo.id} disconnected`);
+        deleteClientById(socket.id)
+    });
+});
+
+io.engine.on("connection_error", (err) => {
+    console.log(err.req);      // the request object
+    console.log(err.code);     // the error code, for example 1
+    console.log(err.message);  // the error message, for example "Session ID unknown"
+    console.log(err.context);  // some additional error context
+  });
+
+// Start the server
+const PORT = process.env.PORT || 3000;
+server.listen(PORT, () => {
+    console.log(`Server listening on port ${PORT}`);
+});
+
+
+
+// Utils
+// Function to delete an item by its id (mutating the array)
+function deleteClientById(id) {
+    const index = clients.findIndex(item => item.id === id);
+    if (index !== -1) {
+        clients.splice(index, 1);
+    }
+}
+
+function returnResponse(request: BaseMessage): Observable<BaseMessage> {
+    return new Observable((observer) => {
+        prepareResponseMessages(10, 1000).subscribe({
+            next: (message: BaseMessage) => {
+                message.header.messageID = request.header.messageID
+                observer.next(message)
+            },
+            error: err => console.error(err),
+            complete: () => {
+                prepareResponseMessages(1).subscribe({
+                    next: message => {
+                        message.header.messageID = request.header.messageID
+                        message.header.messageName = 'Complete'
+                        observer.next(message)
+                    },
+                    complete: () => {
+                        observer.complete()
+                    }
+                })
+            }
+        })
+    })
+}
+
+export interface ClientInfo {
+    id: string,
+    connectedAt: Date,
+    clientConnectionState: BehaviorSubject<'ONLINE' | 'OFFLINE'>,
+    requests: { message: any, completed: boolean }[],
+    buffer: RetransmissionService
+}

test/socket.test.ts

@@ -0,0 +1,14 @@
+// import { receiverConnectionState } from './socket'
+
+// test('should simulate client connectivity', () => {
+//     const expectedValues = ['OFFLINE', 'ONLINE', 'OFFLINE', 'ONLINE'];
+//     let index = 0;
+//     receiverConnectionState.subscribe({
+//         next: (value) => {
+//             expect(value).toBe(expectedValues[index++])
+//         },
+//         complete() {
+//             // nothing
+//         },
+//     })
+// })

test/socket.ts

@@ -0,0 +1,123 @@
+import { Observable, Subject, takeWhile } from "rxjs";
+import { prepareResponseMessages } from "../services/utility/prepareFISmessage";
+import { BaseMessage } from "../dependencies/logging/interface/export";
+import { WrappedMessage } from "../services/retransmission.service";
+import { io, Socket } from "socket.io-client";
+
+let onHoldMessagesSubject: Subject<WrappedMessage> = new Subject()
+let toBePassedOverToApp: Subject<BaseMessage> = new Subject()
+// Serve static files (optional)
+let sender: Subject<BaseMessage> = prepareResponseMessages(1, 2000)
+let serverSocketUrl: string = 'http://localhost:3000'
+let socket: Socket
+establishSocketConnection(serverSocketUrl)
+// interval(1000).subscribe(value => { // just to test if the emission is in sequence after reconnection
+//     console.log(value)
+//     socket.emit('interval', value)
+// })
+sender.subscribe({
+    next: message => {
+        makeRequest(message).subscribe({
+            complete: () => console.log(`Request ${message.header.messageID} has acquired all responses.`)
+        })
+    }
+})
+
+// the interface the client Program will make without having to decide transport protocol
+function makeRequest(request: BaseMessage): Observable<any> {
+    return new Observable((response) => {
+        sendMessage(request)
+        toBePassedOverToApp.subscribe({
+            next: (message: BaseMessage) => {
+                // The identification of responses mapping to the request be adjusted accordingly
+                // For now it's a simple demulti-plexing
+                if (message.header.messageID == request.header.messageID && message.header.messageName == 'ResponseData') {
+                    response.next(message)
+                }
+                if (message.header.messageID == request.header.messageID && message.header.messageName == 'Complete') {
+                    response.complete()
+                }
+            },
+            error: err => console.error(err),
+            complete: () => { }
+        })
+    })
+}
+
+// socket util: Assuming that the client program would already have something like this in place
+function establishSocketConnection(serverUrl: string) {
+    socket = io(serverUrl, {
+        reconnection: true,              // Enable automatic reconnections
+        reconnectionAttempts: 100,        // Retry up to 10 times
+        reconnectionDelay: 500,          // Start with a 500ms delay
+        reconnectionDelayMax: 10000,     // Delay can grow to a max of 10 seconds
+        randomizationFactor: 0.3,
+    })
+
+    // Listen for a connection event
+    socket.on('connect', () => {
+        // socket.emit('Hello from the client!')
+        console.log('Connected to the server:', socket.id)
+        // receiverConnectionState.next('ONLINE')
+    });
+
+    // Listen for messages from the server
+    socket.on('message', (msg: WrappedMessage) => {
+        console.log('Message from server:', msg.payload.header.messageID);
+
+        // Check the sequence by ensuring the message value before the current message exists, then pass them over to "App"
+        onHoldMessagesSubject.next(msg)
+        checkMessage(msg).then(() => [
+            toBePassedOverToApp.next(msg.payload as BaseMessage)
+        ]).catch((err) => console.error(err))
+    })
+
+    socket.on('notification', (msg: string) => {
+        console.log(msg)
+    })
+
+    // Handle disconnection
+    socket.on('disconnect', () => {
+        console.log('Disconnected from the server');
+        // receiverConnectionState.next('OFFLINE')
+    });
+}
+
+
+
+async function sendMessage(message: BaseMessage): Promise<any> {
+    return new Promise((resolve, reject) => {
+        try {
+            // extra precaution: According to chatgpt, if disconnected, then the payload will be loaded back in event queue whilst the socket will try to reestablish connection
+            // https://socket.io/docs/v4/client-offline-behavior/
+            socket.emit('message', message); // inherently an aysnc
+            console.log(`SocketEmit() for message to event queue ${message.header.messageID}`)
+            resolve('')
+        } catch (error) {
+            console.error('Error emitting message:', error);
+            this.wrappedMessage.next(message)
+            reject(error)
+        } ``
+    })
+}
+
+
+// SO concept will be that if the message behind it is received, then 
+async function checkMessage(message: WrappedMessage): Promise<any> {
+    return new Promise((resolve, reject) => {
+        if (message.previousMessageID) {
+            onHoldMessagesSubject.pipe(
+                takeWhile(item => item.payload.header.messageID == message.previousMessageID)
+            ).subscribe({
+                complete: () => {
+                    resolve('previousMessageID matched')
+                    // console.log(`${message.payload.header.messageID} : Previous messageID(${message.previousMessageID}) matched`)
+                    console.log(`matched`)
+                }
+            })
+        } else {
+            console.log('No previous messageID. This should be the first message')
+            resolve('No previous message ID. Please Proceed.')
+        }
+    })
+}

test/socketTest.txt

@@ -0,0 +1,250 @@
+import { BehaviorSubject, buffer, concatMap, distinctUntilChanged, from, interval, Observable, Subject, Subscriber, take, takeUntil, takeWhile } from "rxjs";
+import { io, Socket } from "socket.io-client";
+import { prepareResponseMessages } from "../services/utility/prepareFISmessage";
+import { BaseMessage } from "../dependencies/logging/services/logging-service";
+import { rejects } from "assert";
+// Connect to the server
+const socket: Socket = io('http://localhost:3000');
+
+export let abstractStorage: WrappedMessage[] = []
+export let bufferReleaseSignal: Subject<void> = new Subject()
+export let sender: Subject<BaseMessage> = prepareResponseMessages(3000, 500)
+export let receiverConnectionState: BehaviorSubject<'OFFLINE' | 'ONLINE'> = new BehaviorSubject('OFFLINE')
+export let transmissionState: BehaviorSubject<'TRANSMITTING' | 'IDLE' | 'ARRAY EMPTY' | 'STORING DATA' | 'GETTING STORED DATA'> = new BehaviorSubject('ARRAY EMPTY')
+export let arrayToBeTransmitted: Subject<WrappedMessage[]> = new Subject()
+export let toBeWrapped: Subject<any> = new Subject()
+export let wrappedMessage: Subject<WrappedMessage> = new Subject() 
+export let transportLayerMessages = new Subject<any>()
+
+// run this to active the release mechanism
+releaseSignalManager()
+// sender goes to toBeWrapped
+sender.subscribe(message => toBeWrapped.next(message))
+// toBeWrapped will wrap the message with timeReceived and push next to wrappedMesasge subject
+let currentMessageId: string | null
+toBeWrapped.subscribe(message => {
+    wrappedMessage.next(wrapMessageWithTimeReceived(message, currentMessageId ? currentMessageId : null))
+    currentMessageId = message.header.messageID
+})
+//simulate connection test
+
+// wrappedMessage will then be pushed to buffer
+wrappedMessage.pipe(buffer(bufferReleaseSignal)).subscribe((bufferedMessages: WrappedMessage[]) => {
+    console.log(bufferedMessages.length + ' buffered messages')
+    console.log(`Released buffered message: ${bufferedMessages.length} total messages. To Be sorted.`)
+    // arrayToBeTransmitted.next(sortMessage(bufferedMessages))
+    arrayToBeTransmitted.next(bufferedMessages.length > 0 ? sortMessage(bufferedMessages) : [])
+});
+
+arrayToBeTransmitted.subscribe(array => {
+    if (array.length > 0) {
+        /* Note: Latest update, no point checking the receiver's connection state, since, once the message is pass on, it will
+        be flushed into the event queue to be executed at a later time, which the connnection state would be mutated by then. */
+        // update transmission to indicate that this batch of array is being processed
+        transmissionState.next('TRANSMITTING')
+        from(array).pipe(
+            concatMap((message: WrappedMessage) => {
+                if (transmissionState.getValue() === 'TRANSMITTING') {
+                    console.log(message.timeReceived);
+                    return sendMessage(message).catch((error) => {
+                        return storeMessage(message).then((msgId) => {
+                            console.log(`Message (${msgId}) stored Successfully. {TransmissionState: ${transmissionState.getValue()}}`);
+                        }).catch((error) => {
+                            console.error(error);
+                        });
+                    });
+                } else if (transmissionState.getValue() === 'STORING DATA') {
+                    return storeMessage(message).then((msgId) => {
+                        console.log(`Message (${msgId}) stored Successfully. {TransmissionState: ${transmissionState.getValue()}}`);
+                    }).catch((error) => {
+                        console.error(error);
+                    });
+                } else if (receiverConnectionState.getValue() === 'OFFLINE') {
+                    transmissionState.next('STORING DATA'); // to be fired every message processing
+                    return storeMessage(message).then((msgId) => {
+                        console.log(`Message (${msgId}) stored Successfully. {TransmissionState: ${transmissionState.getValue()}}`);
+                    }).catch((error) => {
+                        console.error(error);
+                    });
+                } else {
+                    return Promise.resolve(); // No async work, but need to return a resolved promise
+                }
+            })
+        ).subscribe({
+            error: err => console.error(err),
+            complete: () => {
+                // update transmission state to indicate this batch is completed
+                console.log(`Processing buffered array completed. Changing transmission state to ARRAY EMPTY`);
+                transmissionState.next('ARRAY EMPTY');
+
+                if (receiverConnectionState.getValue() === 'ONLINE' && transmissionState.getValue() === 'ARRAY EMPTY') {
+                    setTimeout(() => {
+                        bufferReleaseSignal.next()
+                    }, 1000)
+                }
+                // Do nothing if the receiver connection is offline
+            }
+        });
+    } else {
+        // If I don't do setTimeout, then bufferrelasesignal will be overloaded
+        setTimeout(() => {
+            bufferReleaseSignal.next()
+        }, 3000)
+    }
+}
+)
+
+
+/* Utils */
+function releaseSignalManager() {
+    receiverConnectionState.pipe(
+        distinctUntilChanged()
+    ).subscribe(clientState => {
+        console.log(`Client is now ${clientState}`)
+        if (clientState == 'OFFLINE') {
+            console.log(`Current transmission state: ${transmissionState.getValue()}`)
+            // just keep buffering
+        }
+        if (clientState == 'ONLINE') {
+            console.log(`Current transmission state: ${transmissionState.getValue()}`)
+            // get the stored messages to pump it back into the buffer to be ready to be processed immediately
+            if (transmissionState.getValue() == 'ARRAY EMPTY') {
+                getDataAndUpdateState()
+            }
+            if (transmissionState.getValue() == 'STORING DATA') {
+                // have to wait for storing data to be completed before proceeding to the code above
+                transmissionState.pipe(
+                    takeWhile(value => value == 'ARRAY EMPTY') //listen to this value and then destroy this observable
+                ).subscribe({
+                    next: () => {
+                        getDataAndUpdateState()
+                    },
+                    error: err => console.error(err),
+                    complete: () => { }
+                })
+            }
+        }
+    })
+}
+
+function sortMessage(array: WrappedMessage[]): WrappedMessage[] {
+    console.log(`Sorting ${array.length} messages....`)
+    return array.sort((a, b) => {
+        return new Date(a.timeReceived).getTime() - new Date(b.timeReceived).getTime();
+    });
+}
+
+function wrapMessageWithTimeReceived(message: any, previousMessageID: string): any {
+    // check if message has already a time received property if so no need to add anymore
+    if (!message.timeReceived) {
+        let WrappedMessage: WrappedMessage = {
+            timeReceived: new Date(),
+            payload: message as BaseMessage,
+            previousMessageID: previousMessageID
+        }
+        return WrappedMessage
+    } else {
+        return message as WrappedMessage
+    }
+}
+
+async function getStoredMessages(): Promise<WrappedMessage[]> {
+    return new Promise((resolve, reject) => {
+        let array = []
+        setTimeout(() => {
+            abstractStorage.forEach(message => {
+                array.push(message)
+            })
+            abstractStorage = []
+        }, 5000)
+        resolve(array)
+    })
+}
+
+// just an abstraction
+async function storeMessage(message: WrappedMessage): Promise<any> {
+    return new Promise((resolve, reject) => {
+        try {
+            setTimeout(() => {
+                console.log(`Storing ${message.payload.header.messageID}....`)
+                abstractStorage.push(message)
+                resolve(message.payload.header.messageID)
+            }, 1000)
+        }
+        catch (error) {
+            reject(error)
+        }
+    })
+}
+
+function getDataAndUpdateState() {
+    transmissionState.next('GETTING STORED DATA')
+    console.log(`Current transmission state: ${transmissionState.getValue()}`)
+    getStoredMessages().then((storedMessages: WrappedMessage[]) => {
+        if (storedMessages.length > 0) {
+            console.log(`${storedMessages.length} STORED messages.`)
+            from(storedMessages).subscribe({
+                next: message => {
+                    wrappedMessage.next(message)
+                },
+                error: err => console.error(err),
+                complete: () => {
+                    console.log(`Flushed back ${storedMessages.length} messages back in buffer`)
+                    transmissionState.next('ARRAY EMPTY')
+                    bufferReleaseSignal.next()
+                }
+            })
+        } else {
+            console.log(`${storedMessages.length} STORED messages.`)
+            transmissionState.next('ARRAY EMPTY')
+            bufferReleaseSignal.next()
+        }
+    }).catch((err) => {
+        console.error(err)
+    })
+}
+
+export interface WrappedMessage {
+    timeReceived: any, // this property is for sender to sort
+    payload: BaseMessage,
+    previousMessageID?: string // this property is for receiver to sort
+}
+// Listen for a connection event
+socket.on('connect', () => {
+    socket.emit('Hello from the client!')
+    console.log('Connected to the server:', socket.id)
+    receiverConnectionState.next('ONLINE')
+});
+
+// Listen for messages from the server
+socket.on('message', (msg: string) => {
+    console.log('Message from server:', msg);
+})
+
+async function sendMessage(message: WrappedMessage): Promise<any> {
+    return new Promise((resolve, reject) => {
+        try {
+            // extra precaution: According to chatgpt, if disconnected, then the payload will be loaded back in event queue whilst the socket will try to reestablish connection
+            // https://socket.io/docs/v4/client-offline-behavior/
+            socket.emit('message', message); // inherently an aysnc
+            console.log(`SocketEmit() for message to event queue ${message.payload.header.messageID} 
+                current tranmission State: ${transmissionState.getValue()} 
+                current connection State: ${receiverConnectionState.getValue()}
+                ${receiverConnectionState.getValue()=='OFFLINE'? 'Message in the event queue will be attempted again after connection is back' : 'Sent over'}`);
+            resolve('')
+        } catch (error) {
+            console.error('Error emitting message:', error);
+            wrappedMessage.next(message)
+            reject(error)
+        }``
+    })
+}
+
+// Handle disconnection
+socket.on('disconnect', () => {
+    console.log('Disconnected from the server');
+    receiverConnectionState.next('OFFLINE')
+});
+
+
+

test/stringtest.ts

@@ -0,0 +1,44 @@
+const fs = require('fs')
+
+let fp1: string = '' // data
+let fp2: string = '' // output
+// let fp3: string = '' // fingerprint
+
+fs.readFile('data.txt', (err, fingerprint) => {
+    if (err) throw err;
+    fp1 = fingerprint.toString()
+    // console.log(fp1.length);
+})
+fs.readFile('fingerprint.txt', (err, fingerprint) => {
+    if (err) throw err;
+    fp2 = fingerprint.toString()
+    // console.log(fp2.length);
+})
+// fs.readFile('fingerprint.txt', (err, fingerprint) => {
+//     if (err) throw err;
+//     fp3 = fingerprint.toString()
+//     // console.log(fp3);
+// })
+
+const mismatches: number = compareAndCountDifferences(fp1, fp2);
+console.log(`Number of mismatches: ${mismatches}`); 
+
+function compareAndCountDifferences(str1, str2): number {
+    // Length validation:
+    if (str1.length !== 151440 || str2.length !== 151440) {
+      console.log("Error: Strings must be exactly 151440 characters long")
+    }
+  
+    // Character comparison and difference counting:
+    let differenceCount = 0;
+    for (let i = 0; i < str1.length; i++) {
+      if (str1[i] !== str2[i]) {
+        differenceCount++;
+      }
+    }
+  
+    return differenceCount;
+  }
+
+
+

tsconfig.json

@@ -66,7 +66,7 @@
     // "preserveSymlinks": true,                         /* Disable resolving symlinks to their realpath. This correlates to the same flag in node. */
     //"forceConsistentCasingInFileNames": true,            /* Ensure that casing is correct in imports. */
     /* Type Checking */
-    "strict": true, /* Enable all strict type-checking options. */
+    "strict": false, /* Enable all strict type-checking options. */
     "noImplicitAny": false, /* Enable error reporting for expressions and declarations with an implied `any` type.. */
     // "strictNullChecks": true,                         /* When type checking, take into account `null` and `undefined`. */
     // "strictFunctionTypes": true,                      /* When assigning functions, check to ensure parameters and the return values are subtype-compatible. */