fixed the interfaces and the types

doc/explanation.txt

@@ -9,10 +9,6 @@ Things to do:
 
 
 
-
-
-
-
 Concept as of 5/12/2024:
 So, when an application run, it will instantiate the transmission manager that will in turn instantiate adapter manager. The adapter manager will turn first check the 
 configuration file, whether a config json file or .env depending on the environment, and set up the necessary transport services in order to be able to create the 

logSetting.json

@@ -1,5 +1,5 @@
 {
-    "base": false,
+    "base": true,
     "managers": true,
     "transmission": true,
     "adapter": true,

src/actor/transmission.actor.ts

@@ -12,8 +12,6 @@ export class TransmissionActor extends ActorBase<ActorMessage<FisMessage>> {
     constructor(actorParam: ActorProfile<TransmisisonProfile>) {
         super()
         this.setup(actorParam)
-        this.handleMessageTransmission(this.outgoingBus.asObservable())
-        this.handleMessageReception(this.incomingBus)
     }
 
     protected handleMessageTransmission(messageToBeTransmitted: Observable<ActorMessage<FisMessage>>): void {

src/adapters/adapter.manager.ts

@@ -1,57 +1,93 @@
-import { filter, Observable, Observer, Subject } from "rxjs"
-import config from '../config/config.json';
+import { filter, Observable, Observer, Subject, Subscription } from "rxjs"
 import { v4 as uuidv4 } from 'uuid'
 import ConsoleLogger from "../utils/log.utils"
-import { AdapterInterface, ClientObject, GeneralEvent, ReceiverAdapterInterface, TransmitterAdapterInterface, Transport, TransportService, TransportSet } from "../interface/interface"
+import { AdapterInterface, ClientObject, GeneralEvent, ReceiverAdapterInterface, TransmitterAdapterInterface, Transport, TransportServiceInterface } from "../interface/interface"
 import { TransmitterAdapter } from "./adapter.transmitter";
 import { ReceiverAdapter } from "./adapter.receiver";
 import { AdapterManagerBase } from "../base/adapter.manager.base";
 
-/* Note: There will be a need to use the logic in place for socket utility. Especially for client identification
-Will think about that later. Because that mechanism needs to be made universal somehow. If let's say utilizing already
-existing transport, there are no logic in place to exchange information to identify connected clients. */
-
 export class AdapterManager extends AdapterManagerBase {
     private console: ConsoleLogger = new ConsoleLogger(`AdapterManager`, ['managers'])
 
     constructor(event: Subject<GeneralEvent<any>>, browserEnv?: boolean) {
         super()
+        this.browserEnv = browserEnv ?? false
         this.console.log({ message: `Contructing self...` })
         this.event = event
 
+        this.connectToExistingTransport(this.event)
     }
 
-    public subscribeForAdapters(): Observable<AdapterInterface> {
-        return new Observable((observer: Observer<AdapterInterface>) => {
-            this.event.pipe(
-                // usually a new client is notified from the transport service.
-                filter(event => event.event === `New Client`)
-            ).subscribe({
-                next: (event: GeneralEvent<any>) => {
-                    this.handleTransportEvent(event, this.event, event.transport)                                              
+    public subscribeForAdapters(): Observable<GeneralEvent<AdapterInterface>> {
+        return new Observable((observer: Observer<GeneralEvent<AdapterInterface>>) => {
+            const subscription: Subscription = this.event.pipe(
+                filter((event: GeneralEvent<ClientObject>) => event.event === 'New Client')
+            ).subscribe((event: GeneralEvent<ClientObject>) => {
+                let adapters: AdapterInterface[] = this.instantiateAdapterComponents(event.data)
+                if (adapters.length > 0) {
+                    adapters.forEach((adapter: AdapterInterface) => {
+                        observer.next({
+                            id: uuidv4(),
+                            type: `Adapter Event`,
+                            event: `New Adapter`,
+                            transport: event.data.transport,
+                            date: new Date(),
+                            data: adapter
+                        })
+                    })
+                } else {
+                    // handler for no adapters instantiated for this new client
                 }
             })
+
+            return () => {
+                subscription.unsubscribe()
+            }
         })
     }
 
-    private handleTransportEvent(event: GeneralEvent<ClientObject>, adapterEvent: Subject<GeneralEvent<any>>, transport: Transport): void {
-        let transportService: TransportService | undefined = this.transportServiceArray.find(obj => obj.getInfo() === transport)
-        if (transportService) {
-            let transmitterAdapter: TransmitterAdapterInterface = new TransmitterAdapter(event.data.clientId, event.transport, transportService)
-            let receiverAdapter: ReceiverAdapterInterface = new ReceiverAdapter(event.data.clientId, event.transport, transportService)
-            adapterEvent.next({
-                id: uuidv4(),
-                event: 'New Adapter',
-                type: 'Adapter Event',
-                date: new Date(),
-                data: [transmitterAdapter, receiverAdapter]
-            } as GeneralEvent<AdapterInterface[]>)
+    private connectToExistingTransport(event: Subject<GeneralEvent<any>>): void {
+        this.event.pipe(
+            filter(event => event.type === `Transport Event`),
+            filter(event => event.event === `Available Transport`)
+        ).subscribe((event: GeneralEvent<TransportServiceInterface[]>) => {
+            let transportServices: TransportServiceInterface[] = event.data
+            transportServices.forEach((transportService: TransportServiceInterface) => {
+                this.updateTransportServicesRecord(transportService)
+            })
+        })
+        this.event.next({
+            id: uuidv4(),
+            type: 'Adapter Event',
+            event: `Adapter Manager Started`,
+            date: new Date(),
+            data: {
+                event: event
+            }
+        })
+    }
+
+    private updateTransportServicesRecord(transportService: TransportServiceInterface): void {
+        if (this.transportServiceArray.some(obj => obj.getInfo().transportServiceId === transportService.getInfo().transportServiceId)) {
+            this.console.error({ message: `Transport Service Id: ${transportService.getInfo().transportServiceId} already existed.` })
         } else {
-            this.console.error({ message: `No ${transport} service is not properly instantiated....` })
-            throw new Error(`No ${transport} service is not properly instantiated....`)
+            this.transportServiceArray.push(transportService)
+            this.console.log({ message: `Transport Service Id: ${transportService.getInfo().transportServiceId} has been added to adapter's record.` })
         }
     }
 
+    private instantiateAdapterComponents(clientRef: ClientObject): AdapterInterface[] {
+        let transportService = this.transportServiceArray.find(obj => obj.getInfo().transportServiceId === clientRef.transportServiceId)
+        if (transportService) {
+            let transmitterAdapter: TransmitterAdapterInterface = new TransmitterAdapter(clientRef.clientId, transportService.getInfo().transport, transportService)
+            let receiverAdapter: ReceiverAdapterInterface = new ReceiverAdapter(clientRef.clientId, transportService.getInfo().transport, transportService)
+            return [transmitterAdapter, receiverAdapter]
+        } else {
+            this.console.error({ message: `Transport Service id ${clientRef.transportServiceId} not found. Unable to instantiate adapters.` })
+            return []
+        }
+
+    }
 }
 
 

src/adapters/adapter.receiver.ts

@@ -23,7 +23,7 @@ export class ReceiverAdapter extends AdapterBase {
     subscribeForIncoming(): Observable<GeneralEvent<any>> {
         this.console.log({ message: `Connector getting message bus for this connector: ${this.adapterID}` })
         return new Observable((observable: Observer<GeneralEvent<any>>) => {
-            const subscription: Subscription = this.transportService.subscribe().pipe(
+            const subscription: Subscription = this.transportService.subscribeForTransportEvent().pipe(
                 filter((message: GeneralEvent<any>) => message.event === 'New Message'),
                 // take message only specific for this adapter. Although that itself wouldn't be necessary, considerng everything goes through transportEvent. I guess it's for better management
                 filter((message: GeneralEvent<TransportMessage>) => (message.data as TransportMessage).target == this.adapterID),

src/base/adapter.base.ts

@@ -1,6 +1,6 @@
 import { Observable, Subject } from "rxjs";
 import dotenv from 'dotenv';
-import { AdapterInterface, TransmissionRole, Transport, TransportService } from "../interface/interface";
+import { AdapterInterface, TransmissionRole, Transport, TransportServiceInterface } from "../interface/interface";
 
 dotenv.config();
 /* This transport manager will be instantiating the necessary transport to deal with tranmission and receiving from different receivers
@@ -10,7 +10,7 @@ export class AdapterBase implements AdapterInterface {
     role!: TransmissionRole
     transport!: Transport
 
-    protected transportService!: TransportService
+    protected transportService!: TransportServiceInterface
 
     constructor() {
         //logic here

src/base/adapter.manager.base.ts

@@ -1,16 +1,17 @@
 import { Observable, Subject } from "rxjs"
-import { AdapterInterface, AdapterManagerInterface, GeneralEvent, TransportService } from "../interface/interface"
+import { AdapterInterface, AdapterManagerInterface, GeneralEvent, TransportServiceInterface } from "../interface/interface"
 
 export class AdapterManagerBase implements AdapterManagerInterface {
+    protected browserEnv!: boolean
     protected event!: Subject<GeneralEvent<any>>
-    protected transportServiceArray: TransportService[] = []
+    protected transportServiceArray: TransportServiceInterface[] = []
     protected adapters: AdapterInterface[] = []
 
     constructor() {
         // logic here
     }
 
-    subscribeForAdapters(): Observable<AdapterInterface> {
+    subscribeForAdapters(): Observable<GeneralEvent<any>> {
         throw new Error("Method not implemented.")
     }
 

src/base/msg.transmission.base.ts

@@ -1,10 +1,10 @@
-import { Observable } from 'rxjs';
-import { AdapterEvent, AdapterInterface, EventObject, MessageTransmissionInterface } from '../interface/interface'
+import { Observable, Subject } from 'rxjs';
+import { AdapterInterface,  GeneralEvent,  MessageTransmissionInterface } from '../interface/interface'
 
 export class MessageTransmissionBase implements MessageTransmissionInterface {
     protected clientId!: string;
     protected adapters: AdapterInterface[] = []
-    protected eventObj!: EventObject
+    protected event!: Subject<GeneralEvent<any>>
 
     constructor() {
         // logic here

src/base/msg.transmission.manager.base.ts

@@ -1,6 +1,6 @@
 
 import { Observable, Observer, Subject, Unsubscribable } from 'rxjs';
-import { MessageTransmissionManagerInterface, TransmissionInterface } from '../interface/interface';
+import { GeneralEvent, MessageTransmissionManagerInterface, TransmissionInterface } from '../interface/interface';
 import { ActorInterface, ActorProfile } from '../interface/actor.sample';
 import { ActorBase } from './actor.base';
 import { AdapterManager } from '../adapters/adapter.manager';
@@ -9,11 +9,12 @@ export class MessageTransmissionManagerBase implements MessageTransmissionManage
     protected browserEnv!: boolean
     protected transmissionSet: TransmissionInterface[] = []
     protected adapterManager!: AdapterManager
+    protected event!: Subject<GeneralEvent<any>>
 
     constructor() {
         // logic here
     }
-    subscribeForTransmission(): Observable<TransmissionInterface> {
+    subscribeForTransmission(): Observable<GeneralEvent<any>> {
         throw new Error('Method not implemented.');
     }
 

src/interface/interface.ts

@@ -13,11 +13,11 @@ export interface GeneralEvent<T> {
 
 /* MANAGEMENT */
 export interface MessageTransmissionManagerInterface {
-    subscribeForTransmission(): Observable<TransmissionInterface>
+    subscribeForTransmission(): Observable<GeneralEvent<any>>
 }
 
 export interface AdapterManagerInterface {
-    subscribeForAdapters(): Observable<AdapterInterface>
+    subscribeForAdapters(): Observable<GeneralEvent<any>>
 }
 
 
@@ -103,15 +103,20 @@ export enum AdapterTransmissionRole {
 
 export type EventType = `General Event` | 'Transport Event' | 'Transmission Event' | 'Adapter Event'
 export type TransmissionRole = `Transmitter` | 'Receiver' | 'RequestResponse'
-export type EventMessage = 'Server Started' | 'New Client' | 'Client Connected' | 'Client Re-connected' | 'Client Disconnected' | `Server Disconnected` | 'New Message' | `Notification` | `New Server` | `Server Connected` | `New Transport` | 'New Adapter' | 'Re-Flush' | 'New Transport' | 'New Transmission'
+export type EventMessage = 'Server Started' | 'New Client' | 'Client Connected' | 'Client Re-connected' | 'Client Disconnected' | `Server Disconnected` | 'New Message' | `Notification` | `New Server` | `Server Connected` | `New Transport` | 'New Adapter' | 'Re-Flush' | 'New Transport' | 'New Transmission' | 'Adapter Manager Started' | 'Available Transport'
 export type ConnectionState = 'ONLINE' | 'OFFLINE'
 
 export interface TransportServiceInterface {
-    getInfo(): Transport
+    getInfo(): TransportServiceProfile
     emit(message: TransportMessage): void
     subscribeForTransportEvent(): Observable<GeneralEvent<any>> //all messages and whatever event will go through this, easier to implemnet across different transport protocol
 }
 
+export interface TransportServiceProfile {
+    transportServiceId: string,
+    transport: Transport
+}
+
 export interface Info {
     transport: Transport
 }
@@ -119,7 +124,9 @@ export interface Info {
 export interface ClientObject {
     clientId: string,
     dateCreated: Date,
-    connectionState: BehaviorSubject<ConnectionState>
+    connectionState: BehaviorSubject<ConnectionState>,
+    transport: Transport,
+    transportServiceId: string 
 }
 
 

src/test/proxy.ts

@@ -9,12 +9,12 @@ import axios from 'axios';
 let fromServer = new Subject<{ event: 'profile' | 'message', payload: any }>()
 let toServer = new Subject<{ event: 'profile' | 'message', payload: any }>()
 
-// startSocketServer(3001)
+startSocketServer(3001)
 // startSocketServer(3002)
 // startClientSocketConnection('http://localhost:3000')
-startHttpServer(3001).then((app: Express) => {
-    operateHttpServer(app, 'http://localhost:3000/')
-})
+// startHttpServer(3001).then((app: Express) => {
+//     operateHttpServer(app, 'http://localhost:3000/')
+// })
 consoleLog()
 
 function consoleLog(): void {

src/test/receiver.ts

@@ -1,36 +1,47 @@
-import { filter, interval, map, Observable, Observer, Subject } from "rxjs";
+import { filter, interval, map, Observable, Observer, Subject, Subscription } from "rxjs";
 import { v4 as uuidv4 } from 'uuid'
+import config from '../config/config.json';
 import { MessageTransmissionManager } from "../transmission/msg.transmission.manager";
 import { WrappedMessage } from "../utils/message.ordering";
 import ConsoleLogger from "../utils/log.utils";
-import { FisMessage, Transmission, TransportEvent, TransportMessage } from "../interface/interface";
+import { FisMessage, GeneralEvent, TransmissionInterface, Transport, TransportMessage, TransportServiceInterface, TransportSet } from "../interface/interface";
+import { WebsocketTransportService } from "../transport/websocket";
+import { HttpTransportService } from "../transport/http";
 
 class Supervisor {
-    private generalBus: Subject<TransportEvent> = new Subject()
+    private generalBus: Subject<GeneralEvent<any>> = new Subject()
     private console = new ConsoleLogger('Supervisor', ['base'])
     private isClient: boolean = true
     private transmissionManager!: MessageTransmissionManager
-    private event: Observable<TransportEvent>
-    private transmissionSets: Transmission[] = []
+    private event: Subject<GeneralEvent<any>>
+    private transmissionSets: TransmissionInterface[] = []
     private outgoingPipe: Subject<any> = new Subject()
+    private transportServiceArray: TransportServiceInterface[] = []
+    private transportSet: TransportSet[] = []
 
     constructor() {
-        this.transmissionManager = new MessageTransmissionManager(this.isClient)
-        this.event = this.transmissionManager.getEvent()
+        this.event = new Subject()
+        this.transmissionManager = new MessageTransmissionManager(this.event, this.isClient)
+        this.sortTransportFromEnv(this.transportSet)
+        this.updateAdapterManagerForTransport(this.event)
+        this.transportSet.forEach(transport => {
+            this.setUpTransportService(transport, this.event, this.isClient)
+        })
 
-        this.transmissionManager.subscribe().subscribe((transmissionSet: Transmission) => {
-            this.console.log({ message: `Acquired transmission set for client` })
-            this.transmissionSets.push(transmissionSet)
+        this.transmissionManager.subscribeForTransmission().subscribe((event: GeneralEvent<TransmissionInterface>) => {
+            let transmission: TransmissionInterface = event.data
+            this.console.log({ message: `Acquired transmission set for client ${transmission.clientId}` })
+            this.transmissionSets.push(transmission)
 
-            this.handleActivity(transmissionSet)
-            this.outgoingPipe.subscribe(message => transmissionSet.transmitter.emit(message))
+            this.handleActivity(transmission)
+            this.outgoingPipe.subscribe(message => transmission.transmitter.emit(message))
         })
     }
 
     // only called once for each connected clients.
-    private handleActivity(messageTransmission: Transmission): void {
+    private handleActivity(messageTransmission: TransmissionInterface): void {
         // start listening to incoming messages from this client
-        messageTransmission.receiver.subscribe().subscribe((event: TransportEvent) => {
+        messageTransmission.receiver.getIncoming().subscribe((event: GeneralEvent<any>) => {
             this.console.log({ message: `General Bus ${event.event} ${(((event.data as TransportMessage)?.payload as WrappedMessage)?.payload as FisMessage)?.header?.messageID ?? 'Not Message'}`, details: event })
             this.generalBus.next(event)
         })
@@ -51,7 +62,7 @@ class Supervisor {
         // this.startGeneratingRequest(1000, this.outgoingPipe)
     }
 
-    private request(request: FisMessage, messageTransmission: Transmission): Observable<any> {
+    private request(request: FisMessage, messageTransmission: TransmissionInterface): Observable<any> {
         return new Observable((response: Observer<any>) => {
             messageTransmission.transmitter.emit(request)
             this.generalBus.pipe(
@@ -82,6 +93,69 @@ class Supervisor {
         })
     }
 
+
+    // Server to be set up as well as acquiring client information if needed. Like in the case for grpc and socket. Http not requ`ired.
+    private setUpTransportService(transportSet: TransportSet, event: Subject<GeneralEvent<any>>, isClient?: boolean): void {
+        try {
+            let transportService: TransportServiceInterface = this.instantiateTransportService(transportSet.transport, event)
+            this.transportServiceArray.push(transportService)
+            if (transportService instanceof WebsocketTransportService) {
+                this.console.log({ message: `Just Double Checking... this is websocket` })
+                if (isClient) {
+                    // please note this is subject to change depending on the UI environemnt. Angular has their own built in function to read json file based on Swopt-UI
+                    transportService.startClient(config.connection.transmitter)
+                } else {
+                    transportService.startServer(transportSet.port);
+                }
+            } else if (transportService instanceof HttpTransportService) {
+                this.console.log({ message: `Just Double Checking... this is http` })
+                // Additional Http-specific setup if needed.
+                if (isClient) {
+                    transportService.startClient(config.connection.transmitter)
+                } else {
+                    transportService.startServer(transportSet.port)
+                }
+            }
+        } catch (error) {
+            this.console.error({ message: 'Fail to set transport. Error in setting up transport', details: error })
+        }
+    }
+
+    private instantiateTransportService(transportType: Transport, event: Subject<GeneralEvent<any>>): TransportServiceInterface {
+        if (transportType === 'Websocket') {
+            return new WebsocketTransportService(event)
+        }
+        else if (transportType === 'Http') {
+            return new HttpTransportService(event)
+        } else {
+            throw new Error(`No Transport Service Instantiated`)
+        }
+    }
+
+    private sortTransportFromEnv(transportSet: TransportSet[]): void {
+        let transportList: string[] = process.env.Transport?.split(',') || []
+        let portList: number[] = (process.env.PORT?.split(',') || []).map(port => Number(port));
+        transportList.forEach((transport, index) => {
+            transportSet.push({ transport: transport, port: portList[index] } as unknown as TransportSet)
+        })
+        this.console.log({ message: 'TransportSetList', details: this.transportSet })
+    }
+
+    private updateAdapterManagerForTransport(eventBus: Subject<GeneralEvent<any>>): void {
+        const subscription: Subscription = eventBus.pipe(
+            filter(event => event.type === `Adapter Event`),
+            filter(event => event.event === `Adapter Manager Started`)
+        ).subscribe((event: GeneralEvent<any>) => {
+            eventBus.next({
+                id: uuidv4(),
+                type: `General Event`,
+                event: `Available Transport`,
+                date: new Date(),
+                data: this.transportServiceArray
+            })
+        })
+    }
+
 }
 
 let supervisor = new Supervisor()

src/test/transmitter.ts

@@ -3,34 +3,49 @@ import { v4 as uuidv4 } from 'uuid'
 import { MessageTransmissionManager } from "../transmission/msg.transmission.manager";
 import { WrappedMessage } from "../utils/message.ordering";
 import ConsoleLogger from "../utils/log.utils";
-import { FisMessage, GeneralEvent, TransmissionInterface, TransportMessage } from "../interface/interface";
+import { FisMessage, GeneralEvent, TransmissionInterface, Transport, TransportMessage, TransportServiceInterface, TransportSet } from "../interface/interface";
 import config from '../config/config.json';
+import { startSocketServer } from "../utils/socket.utils";
+import { WebsocketTransportService } from "../transport/websocket";
+import { HttpTransportService } from "../transport/http";
 class Supervisor {
     private console = new ConsoleLogger('Supervisor', ['base'])
     private clientIncomingMessage: Subject<FisMessage> = new Subject()
     private messageProducer!: MessageProducer
     private transmissionManager!: MessageTransmissionManager
-    private event!: Observable<GeneralEvent<any>>
+    private event!: Subject<GeneralEvent<any>>
     private transmissionSets: TransmissionInterface[] = []
+    private transportSet: TransportSet[] = []
+    private transportServiceArray: TransportServiceInterface[] = []
 
     constructor() {
+        this.event = new Subject<GeneralEvent<any>>()
+        // set up all the transportServices
+        this.sortTransportFromEnv(this.transportSet)
+        this.transportSet.forEach(set => {
+            this.setUpTransportService(set, this.event)
+        })
         // so need them adapters now. But supervisor shouldn't be concerned, only messageTransmissionManager and ConnectionManager
         this.messageProducer = new MessageProducer(this.clientIncomingMessage)
-        this.transmissionManager = new MessageTransmissionManager()
-        this.event = this.transmissionManager.getEvent()
-
-        this.transmissionManager.subscribeForTransmission().subscribe((transmissionSet: TransmissionInterface) => {
-            this.transmissionSets.push(transmissionSet)
+        this.transmissionManager = new MessageTransmissionManager(this.event)
+        this.startMessageTransmission(this.event)
+    }
 
-            this.handleClientActivity(transmissionSet)
+    private startMessageTransmission(transmissionEvent: Subject<GeneralEvent<TransmissionInterface>>): void {
+        this.transmissionManager.subscribeForTransmission().pipe(
+            filter(event => event.event == `New Transmission`)
+        ).subscribe(event => {
+            // update transmission record on every new transmission object instantiated
+            this.transmissionSets.push(event.data as TransmissionInterface)
+            // start message transmission for said transmission object instantiated
+            this.handleClientActivity(event.data as TransmissionInterface)
         })
-
     }
 
     // only called once for each connected clients.
     private handleClientActivity(messageTransmission: TransmissionInterface): void {
         // start listening to incoming messages from this client
-        messageTransmission.receiver.subscribe().subscribe((event: GeneralEvent<TransportMessage>) => {
+        messageTransmission.receiver.getIncoming().subscribe((event: GeneralEvent<TransportMessage>) => {
             let requestMessage: FisMessage = ((event.data as TransportMessage).payload as WrappedMessage).payload as FisMessage
             this.console.log({ message: `General Bus ${requestMessage?.header?.messageID ?? 'Not a message'}`, details: event }) // receiving end
             this.clientIncomingMessage.next(requestMessage)
@@ -46,6 +61,53 @@ class Supervisor {
             messageTransmission.transmitter.emit(message)
         })
     }
+
+    // Server to be set up as well as acquiring client information if needed. Like in the case for grpc and socket. Http not requ`ired.
+    private setUpTransportService(transportSet: TransportSet, event: Subject<GeneralEvent<any>>, isClient?: boolean): void {
+        try {
+            let transportService: TransportServiceInterface = this.instantiateTransportService(transportSet.transport, event)
+            this.transportServiceArray.push(transportService)
+            if (transportService instanceof WebsocketTransportService) {
+                this.console.log({ message: `Just Double Checking... this is websocket` })
+                if (isClient) {
+                    // please note this is subject to change depending on the UI environemnt. Angular has their own built in function to read json file based on Swopt-UI
+                    transportService.startClient(config.connection.transmitter)
+                } else {
+                    transportService.startServer(transportSet.port);
+                }
+            } else if (transportService instanceof HttpTransportService) {
+                this.console.log({ message: `Just Double Checking... this is http` })
+                // Additional Http-specific setup if needed.
+                if (isClient) {
+                    transportService.startClient(config.connection.transmitter)
+                } else {
+                    transportService.startServer(transportSet.port)
+                }
+            }
+        } catch (error) {
+            this.console.error({ message: 'Fail to set transport. Error in setting up transport', details: error })
+        }
+    }
+
+    private instantiateTransportService(transportType: Transport, event: Subject<GeneralEvent<any>>): TransportServiceInterface {
+        if (transportType === 'Websocket') {
+            return new WebsocketTransportService(event)
+        }
+        else if (transportType === 'Http') {
+            return new HttpTransportService(event)
+        } else {
+            throw new Error(`No Transport Service Instantiated`)
+        }
+    }
+
+    private sortTransportFromEnv(transportSet: TransportSet[]): void {
+        let transportList: string[] = process.env.Transport?.split(',') || []
+        let portList: number[] = (process.env.PORT?.split(',') || []).map(port => Number(port));
+        transportList.forEach((transport, index) => {
+            transportSet.push({ transport: transport, port: portList[index] } as unknown as TransportSet)
+        })
+        this.console.log({ message: 'TransportSetList', details: this.transportSet })
+    }
 }
 
 

src/transmission/msg.transmission.manager.ts

@@ -9,70 +9,65 @@ import { TransmitterAdapter } from "../adapters/adapter.transmitter"
 import { ReceiverAdapter } from "../adapters/adapter.receiver"
 import { MessageTransmissionManagerBase } from "../base/msg.transmission.manager.base";
 import { ActorInterface, ActorProfile } from "../interface/actor.sample";
-import { TransportService } from "../interface/interface";
+import { GeneralEvent, TransmissionInterface, TransportServiceInterface } from "../interface/interface";
 
 export class MessageTransmissionManager extends MessageTransmissionManagerBase {
     private console: ConsoleLogger = new ConsoleLogger(`MessageTransmissionManager`, ['managers'])
 
-    constructor(transportRef: Observable<TransportService>, browserEnv?: boolean) {
+    constructor(event: Subject<GeneralEvent<any>>, browserEnv?: boolean) {
         super()
+        if (browserEnv) this.browserEnv = browserEnv
         this.console.log({ message: `Constructing self...` })
-        transportRef.subscribe((transport: TransportService) => {
-            this.outgoingBus.next({
-                id: uuidv4(),
-                type: `Transport Event`,
-                event: `New Transport`,
-                date: new Date(),
-                data: transport
-            })
+        this.event = event
+        this.event.subscribe((transport: GeneralEvent<TransportServiceInterface>) => {
         })
-        if (browserEnv) this.browserEnv = browserEnv
         // Subscribe for adapterManager and it's relevent event
-        this.adapterManager = new AdapterManager()
+        this.adapterManager = new AdapterManager(event)
         this.adapterManager
     }
 
-    public subscribe(actorProfile: ActorProfile, observer: Partial<Observer<GeneralEvent>>, subscribable?: ActorInterface<GeneralEvent>): Unsubscribable {
-        const targetEvent: EventMessage = this.browserEnv ? 'New Server' : 'New Client';
-        this.eventObj.transportEvent.pipe(
-            filter(event => event.event == targetEvent)
-        ).subscribe(event => {
-            // get all adapters for all the connection
-            let transmission: Transmission | undefined = this.instantiateTransmissionComponents(event?.data?.clientId)
-            if (transmission) {
-                observer.next({
-                    id: uuidv4(),
-                    type: `Transmission Event`,
-                    event: 'New Transmission',
-                    date: new Date(),
-                    transmission: transmission
-                })
-            }
+    public subscribeForTransmission(): Observable<GeneralEvent<TransmissionInterface>> {
+        return new Observable((observer: Observer<GeneralEvent<TransmissionInterface>>) => {
+            this.event.pipe(
+                filter(event => event.event == 'New Client')
+            ).subscribe(event => {
+                // get all adapters for all the connection
+                let transmission: TransmissionInterface | undefined = this.instantiateTransmissionComponents(event?.data?.clientId)
+                if (transmission) {
+                    observer.next({
+                        id: uuidv4(),
+                        type: `Transmission Event`,
+                        event: 'New Transmission',
+                        date: new Date(),
+                        data: transmission
+                    })
+                }
+            })
         })
     }
 
-    protected instantiateTransmissionComponents(clientId: string): Transmission {
-        let receiverInstance: MessageTransmissionReceiver = this.getReceiver(clientId, this.eventObj)
-        let transmitterInstance: MessageTransmissionTransmitter = this.getTransmitter(clientId, this.eventObj)
-        let requestResponseInstance: MessageTransmissionRequestResponse = this.getRequestResponse(clientId, this.eventObj, transmitterInstance, receiverInstance)
-        let transmissionObj: Transmission = {
+    protected instantiateTransmissionComponents(clientId: string): TransmissionInterface {
+        let receiverInstance: MessageTransmissionReceiver = this.getReceiver(clientId, this.event)
+        let transmitterInstance: MessageTransmissionTransmitter = this.getTransmitter(clientId, this.event)
+        let requestResponseInstance: MessageTransmissionRequestResponse = this.getRequestResponse(clientId, this.event, transmitterInstance, receiverInstance)
+        let transmissionObj: TransmissionInterface = {
             clientId: clientId,
             transmitter: transmitterInstance,
             receiver: receiverInstance,
             requestResponse: requestResponseInstance,
-            event: this.eventObj.globalEvent
+            event: this.event
         }
 
         return transmissionObj
     }
 
-    private getReceiver(clientId: string, eventObj: EventObject): MessageTransmissionReceiver {
+    private getReceiver(clientId: string, eventObj: Subject<GeneralEvent<any>>): MessageTransmissionReceiver {
         throw new Error(`Method not defined`)
     }
-    private getTransmitter(clientId: string, eventObj: EventObject): MessageTransmissionTransmitter {
+    private getTransmitter(clientId: string, eventObj: Subject<GeneralEvent<any>>): MessageTransmissionTransmitter {
         throw new Error(`Method not defined`)
     }
-    private getRequestResponse(clientId: string, eventObj: EventObject, transmitter: MessageTransmissionTransmitter, receiver: MessageTransmissionReceiver): MessageTransmissionRequestResponse {
+    private getRequestResponse(clientId: string, eventObj: Subject<GeneralEvent<any>>, transmitter: MessageTransmissionTransmitter, receiver: MessageTransmissionReceiver): MessageTransmissionRequestResponse {
         throw new Error(`Method not defined`)
     }
 

src/transmission/msg.transmission.receiver.ts

@@ -15,9 +15,9 @@ export class MessageTransmissionReceiver extends MessageTransmissionBase impleme
     constructor(clientId: string, event:  Subject<GeneralEvent<any>>) {
         super()
         this.clientId = clientId
-        this.eventObj = event
+        this.event = event
 
-        this.handleAdapterEvent(this.eventObj.adapterEvent.asObservable())
+        this.handleAdapterEvent(this.event.asObservable())
     }
 
     getIncoming(): Observable<GeneralEvent<TransportMessage>> {

src/transmission/msg.transmission.request-response.ts

@@ -1,19 +1,15 @@
 import { MessageTransmissionBase } from "../base/msg.transmission.base";
-import { FisMessage, MessageRequestResponse as MessageRequestResponseInterface } from '../interface/transport.interface'
 import { filter, Observable, Observer, Subject, Subscription, takeWhile } from "rxjs";
 import { v4 as uuidv4 } from 'uuid'
-import { TransportEvent } from "../interface/connector.interface";
 import { MessageTransmissionReceiver } from "./msg.transmission.receiver";
 import { MessageTransmissionTransmitter } from "./msg.transmission.transmitter";
-import { Adapter } from "../base/adapter.base";
+import { AdapterInterface, FisMessage, GeneralEvent, MessageRequestResponseInterface } from "../interface/interface";
 
 export class MessageTransmissionRequestResponse extends MessageTransmissionBase implements MessageRequestResponseInterface {
     transmitterInstance!: MessageTransmissionTransmitter;
     receiverInstance!: MessageTransmissionReceiver;
-    incomingMessageBus!: Subject<any>;
-    outgoingMessageBus!: Subject<any>;
 
-    constructor(transmitterInstance: MessageTransmissionTransmitter, receiverInstance: MessageTransmissionReceiver, event: Observable<TransportEvent>) {
+    constructor(transmitterInstance: MessageTransmissionTransmitter, receiverInstance: MessageTransmissionReceiver, event: Subject<GeneralEvent<any>>) {
         super()
         this.setTransmissionProfile(transmitterInstance, receiverInstance)
         this.event = event
@@ -27,29 +23,11 @@ export class MessageTransmissionRequestResponse extends MessageTransmissionBase
     // To be Enhanced. This is actually wrong at the moment
     send(message: FisMessage): Observable<FisMessage> {
         return new Observable((response: Observer<FisMessage>) => {
-            this.transmitterInstance.emit(message);
-
-            const subscription: Subscription = this.incomingMessageBus
-                .pipe(
-                    // Filter for matching message IDs
-                    filter(incomingMessage => incomingMessage.header.messageID === message.header.messageID),
-                    // Complete the observable when the data is 'Complete'
-                    takeWhile(incomingMessage => incomingMessage.data !== 'Complete')
-                )
-                .subscribe({
-                    next: incomingMessage => {
-                        response.next(incomingMessage);
-                        // Complete the observable if data is 'Complete'
-                        if (incomingMessage.data === 'Complete') {
-                            response.complete();
-                            subscription.unsubscribe();
-                        }
-                    }
-                });
+            // logic here
         });
     }
 
-    setUpAdapter(adapter: Adapter): void {
-        this.mainAdapter = adapter
+    setUpAdapter(adapter: AdapterInterface): void {
+        // logic here
     }
 }

src/transmission/msg.transmission.transmitter.ts

@@ -4,8 +4,7 @@ import { BehaviorSubject, distinct, distinctUntilChanged, filter, map, Observabl
 import { RetransmissionService } from "../utils/retransmission.service";
 import { WrappedMessage } from "../utils/message.ordering";
 import ConsoleLogger from "../utils/log.utils";
-import { TransmitterAdapter } from "../adapters/adapter.transmitter";
-import { AdapterEvent, AdapterInterface, ConnectionState, EventObject, FisMessage, MessageTransmitterInterface, TransmitterAdapterInterface, TransportEvent, TransportMessage } from "../interface/interface";
+import { AdapterInterface, ConnectionState, FisMessage, GeneralEvent, MessageTransmitterInterface, TransmitterAdapterInterface, TransportMessage } from "../interface/interface";
 import { error } from "console";
 
 /* Take in all the messages that needs to be transported, and divide them accordingly. So the connector instances will do just that
@@ -17,13 +16,13 @@ export class MessageTransmissionTransmitter extends MessageTransmissionBase impl
     private buffer!: RetransmissionService;
     private currentAdapter!: TransmitterAdapterInterface
 
-    constructor(clientId: string, eventObj: EventObject) {
+    constructor(clientId: string, event: Subject<GeneralEvent<any>>) {
         super()
         this.console.log({ message: `Constructing Transmitter Transmission with ${clientId}` })
-        this.eventObj = eventObj
+        this.event = event
         this.messageToBeTransmitted = new Subject()
         this.buffer = new RetransmissionService()
-        this.handleAdapters(eventObj.adapterEvent)
+        this.handleAdapters(this.event)
         this.setupBuffer()
 
         // special case just for http in case of server/client disconnected, the unsent msg will be flushed back into messageToBeTransmitted
@@ -37,7 +36,7 @@ export class MessageTransmissionTransmitter extends MessageTransmissionBase impl
 
     private setupBuffer(): void {
         this.console.log({ message: `Setting up Retransmission Service...` })
-        this.eventObj.transportEvent.pipe(
+        this.event.pipe(
             filter(event => event.data.clientId == this.clientId),
             filter(event => event.event == 'Client Disconnected' || event.event == 'Client Re-connected' || event.event == 'Client Connected' || event.event == 'Server Disconnected' || event.event == 'Server Connected'),
             map(event => {
@@ -67,14 +66,14 @@ export class MessageTransmissionTransmitter extends MessageTransmissionBase impl
         })
     }
 
-    private handleAdapters(adaptersEvent: Observable<AdapterEvent>): void {
+    private handleAdapters(adaptersEvent: Observable<GeneralEvent<any>>): void {
         this.handleNewAdapters(adaptersEvent)
     }
 
-    private handleNewAdapters(adaptersEvent: Observable<AdapterEvent>): void {
+    private handleNewAdapters(adaptersEvent: Observable<GeneralEvent<any>>): void {
         adaptersEvent.pipe(
             filter(event => event.event === `New Adapter`),
-            map(event => { return event.adapters }),
+            map(event => { return event.data }),
         ).subscribe({
             next: (adapters: AdapterInterface[]) => {
                 adapters.forEach((adapter: AdapterInterface) => {
@@ -88,7 +87,7 @@ export class MessageTransmissionTransmitter extends MessageTransmissionBase impl
         })
     }
 
-    private handleAdaptersTermination(adaptersEvent: Observable<AdapterEvent>): void {
+    private handleAdaptersTermination(adaptersEvent: Observable<GeneralEvent<any>>): void {
     }
 
     // temporary logic for now. 
@@ -100,11 +99,11 @@ export class MessageTransmissionTransmitter extends MessageTransmissionBase impl
         }
     }
 
-    private uniqueHandlerToFlushUnsentMessages(event: Observable<TransportEvent>): void {
+    private uniqueHandlerToFlushUnsentMessages(event: Observable<GeneralEvent<any>>): void {
         event.pipe(
             filter(event => event.event == 'Re-Flush'),
             filter(event => event.data.clientId == this.clientId),
-        ).subscribe((event: TransportEvent) => {
+        ).subscribe((event: GeneralEvent<any>) => {
             this.console.log({ message: `${this.connectionStateEvent.getValue() == 'ONLINE' ? `Transmitting ${((event.data.payload as TransportMessage).payload as WrappedMessage).thisMessageID}` : `Buffering ${((event.data.payload as TransportMessage).payload as WrappedMessage).thisMessageID}`}` })
             this.messageToBeTransmitted.next(((event.data.payload as TransportMessage).payload as WrappedMessage))
         })

src/transport/gRPC.ts

@@ -1,8 +1,8 @@
 import { Observable } from "rxjs";
-import { TransportEvent } from "../interface/connector.interface";
+import { GeneralEvent } from "../interface/interface";
 
 export class GrpcTransportService {
-    getTransportEventNotification(): Observable<TransportEvent> {
+    getTransportEventNotification(): Observable<GeneralEvent<any>> {
         throw new Error("Method not implemented.");
     }
 

src/transport/http.ts

@@ -7,26 +7,30 @@ import { WrappedMessage } from '../utils/message.ordering';
 import { error } from 'console';
 import axios, { AxiosError } from 'axios';
 import ConsoleLogger from '../utils/log.utils';
-import { ClientObject, GeneralEvent, Transport, TransportMessage, TransportServiceInterface } from '../interface/interface';
+import { ClientObject, GeneralEvent, Transport, TransportMessage, TransportServiceInterface, TransportServiceProfile } from '../interface/interface';
 
 export class HttpTransportService implements TransportServiceInterface {
     private console: ConsoleLogger = new ConsoleLogger(`HttpTransportService`, ['transport'])
     private baseUrl!: string;
-    private info: Transport = `Http`
+    private info!: TransportServiceProfile
     private connectedHttpServer: ConnectedHttpServer[] = [] // to allow the possibility of having to communicate with multiple servers as a client
     private connectedHttpClients: ConnectedHttpClient[] = [] // to keep track of the all the clients that are connected
     transportEvent!: Subject<GeneralEvent<any>>
 
     constructor(event: Subject<GeneralEvent<any>>) {
+        this.info = {
+            transportServiceId: uuidv4(),
+            transport: 'Http'
+        }
         this.baseUrl = config.connection.transmitter
         this.transportEvent = event
     }
 
-    subscribeForTransportEvent(): Observable<GeneralEvent<any>> {
+    public subscribeForTransportEvent(): Observable<GeneralEvent<any>> {
         throw new Error('Method not implemented.');
     }
 
-    public getInfo(): Transport {
+    public getInfo(): TransportServiceProfile {
         return this.info
     }
 
@@ -74,7 +78,7 @@ export class HttpTransportService implements TransportServiceInterface {
     public startServer(port: number): void {
         startHttpServer(port).subscribe({
             next: (client: ConnectedHttpClient) => {
-                handleHttpClient(client, this.connectedHttpClients).subscribe({
+                handleHttpClient(this.info.transportServiceId, client, this.connectedHttpClients).subscribe({
                     next: event => this.transportEvent.next(event),
                     error: error => console.error(error),
                     complete: () => (`Client ${client.clientId} disconnected...`)
@@ -86,7 +90,7 @@ export class HttpTransportService implements TransportServiceInterface {
     }
 
     public startClient(url: string, receiverProfileInfo?: ConnectedHttpServer | undefined): void {
-        initiateClientToServer(url, this.transportEvent, this.connectedHttpServer, receiverProfileInfo).then((connectedHttpServer: ConnectedHttpServer) => {
+        initiateClientToServer(this.info.transportServiceId, url, this.transportEvent, this.connectedHttpServer, receiverProfileInfo).then((connectedHttpServer: ConnectedHttpServer) => {
             handleClientHttpConnection(url, connectedHttpServer).subscribe({
                 next: event => this.transportEvent.next(event),
                 error: error => this.console.log({ message: `Observer Error`, details: error }),

src/transport/websocket.ts

@@ -1,15 +1,16 @@
 import { Observable, Subject } from "rxjs";
+import { v4 as uuidv4 } from 'uuid'
 import { Socket as SocketForConnectedServer } from 'socket.io-client'
 import { Socket as SocketForConnectedClient } from "socket.io"
 import { handleClientSocketConnection, handleNewSocketClient, startClientSocketConnection, startSocketServer } from "../utils/socket.utils";
 import { WrappedMessage } from "../utils/message.ordering";
 import ConsoleLogger from "../utils/log.utils";
-import { ClientObject, FisMessage, GeneralEvent, Transport, TransportMessage, TransportServiceInterface } from "../interface/interface";
+import { ClientObject, FisMessage, GeneralEvent, Transport, TransportMessage, TransportServiceInterface, TransportServiceProfile } from "../interface/interface";
 
 /* Just code in the context that this websocket service will be handling multiple UI clients. Can think about the server communication at a later time. */
 export class WebsocketTransportService implements TransportServiceInterface {
     private console: ConsoleLogger = new ConsoleLogger(`WebsocketTransportService`, ['transport'])
-    private info: Transport = `Websocket`
+    private info!: TransportServiceProfile
     private connectedSocketServer: ConnectedSocketServer[] = [] // to allow the possibility of having to communicate with multiple servers as a client
     private connectedClientSocket: ConnectedSocketClient[] = [] // to keep track of the all the clients that are connected
     // private incomingMessage: Subject<TransportMessage> = new Subject() // this is only for client roles only atm
@@ -17,15 +18,18 @@ export class WebsocketTransportService implements TransportServiceInterface {
 
     constructor(event: Subject<GeneralEvent<any>>) {
         this.console.log({ message: `WebsocketTransportService: Constructing socket transport service....` })
-        this.transportEvent = event
+        this.info = {
+            transportServiceId: uuidv4(),
+            transport: 'Websocket'
+        }
         // logic here
+        this.transportEvent = event
     }
 
     public startServer(port: number): void {
-        // logic here
         startSocketServer(port).subscribe({
             next: (connectedClient: SocketForConnectedClient) => {
-                handleNewSocketClient(connectedClient, this.connectedClientSocket).subscribe({
+                handleNewSocketClient(this.info.transportServiceId, connectedClient, this.connectedClientSocket).subscribe({
                     next: event => this.transportEvent.next(event),
                     error: error => this.console.error({ message: `Observer Error: ${error}`, details: error }),
                     complete: () => this.console.log({ message: `Client ${connectedClient.id} disconnected...` })
@@ -39,7 +43,7 @@ export class WebsocketTransportService implements TransportServiceInterface {
     public startClient(url: string): void {
         // logic here
         startClientSocketConnection(url).then((socket: SocketForConnectedServer) => {
-            handleClientSocketConnection(socket, this.connectedSocketServer).subscribe(this.transportEvent)
+            handleClientSocketConnection(this.info.transportServiceId, socket, this.connectedSocketServer).subscribe(this.transportEvent)
         }).catch((error) => {
             this.console.log({ message: `Observer Error`, details: error })
         })
@@ -65,7 +69,7 @@ export class WebsocketTransportService implements TransportServiceInterface {
         return this.transportEvent.asObservable()
     }
 
-    public getInfo(): Transport {
+    public getInfo(): TransportServiceProfile {
         return this.info
     }
 }

src/utils/general.utils.ts

@@ -0,0 +1,117 @@
+import * as fs from 'fs'
+import path from 'path';
+
+// Check if filename exists. Return profile information if there's any
+export async function checkOwnClientInfo(filename?: string): Promise<{ id: string }> {
+    return new Promise((resolve, reject) => {
+        // Check if the file exists
+        let filePath = process.env.FolderPath as string
+        if (fs.existsSync(`${filePath}${filename}.json`)) {
+            try {
+                // Read the file contents
+                const fileData = fs.readFileSync(filePath + `${filename}.json`, 'utf8');
+
+                // If the file is empty, return an error
+                if (fileData.trim() === "") {
+                    throw new Error("File is empty");
+                }
+
+                // Parse and return the data if present
+                const jsonData = JSON.parse(fileData);
+                resolve(jsonData)
+
+            } catch (err) {
+                // Handle parsing errors or other file-related errors
+                console.log({ message: "Error reading or parsing file:", details: err })
+                reject('');
+            }
+        } else {
+            console.log({ message: `File does not exist. Id: ${filename}` })
+            reject('');
+        }
+    })
+}
+
+export async function checkIfClientExists(id: string): Promise<any> {
+    return new Promise((resolve, reject) => {
+        try {
+            // Check if the file exists
+            let filePath = process.env.FolderPath as string + 'clients.json'
+            if (!fs.existsSync(filePath)) {
+                console.log({ message: `File does not exist.` })
+                reject('File does not exist');
+            }
+
+            // Read and parse the data
+            const fileContent = fs.readFileSync(filePath, 'utf-8');
+            const data: any[] = JSON.parse(fileContent);
+
+            // Check if an details with the given id exists
+            let obj = data.find(entry => entry.id === id);
+
+            if (obj) {
+                console.log({ message: "Client with ID ${id} exists." })
+            } else {
+                console.log({ message: `Client with ID ${id} does not exist.` })
+            }
+            resolve(obj);
+        } catch (error) {
+            reject(`Error reading the file`)
+        }
+    })
+}
+
+export async function writeFile(data: any, filename: string): Promise<boolean> {
+    return new Promise((resolve, reject) => {
+        // Ensure the folder exists
+        const folderPath = process.env.FolderPath as string
+        console.log({ message: folderPath })
+        // const folderPath = path.join(__dirname, folder);
+        if (!fs.existsSync(folderPath)) {
+            fs.mkdirSync(folderPath, { recursive: true }); // Create folder if it doesn't exist
+        } else {
+            console.log({ message: 'Folder already exist' })
+        }
+
+        // Construct the full file path (include the folder)
+        const filePath = path.join(folderPath, `${filename}.json`);
+
+        // Write JSON data to a file
+        fs.writeFile(filePath, JSON.stringify(data, null, 2), (err) => {
+            if (err) {
+                console.log({ message: 'Error writing file', details: err });
+                reject(false);
+            } else {
+                console.log({ message: 'File has been written', details: filePath });
+                resolve(true);
+            }
+        });
+    });
+}
+
+/* For Internal Usage only. Temporary serve as a way for server to keep track of clients. To be replaced in the future with better alternatives. */
+export function addClientToDB(entry: any): void {
+    try {
+        let data: any[] = [];
+        let filePath = process.env.FolderPath as string + 'clients.json'
+        // Check if the file exists and load existing data
+        if (fs.existsSync(filePath)) {
+            const fileContent = fs.readFileSync(filePath, 'utf-8');
+            data = JSON.parse(fileContent);
+        }
+
+        // Append the new details to the array
+        data.push({
+            id: entry.clientId,
+            dateCreated: entry.dateCreated,
+            connectionState: null,
+            socketInstance: null
+        });
+
+        // Write the updated array back to the file
+        fs.writeFileSync(filePath, JSON.stringify(data, null, 2), 'utf-8');
+        console.log({ message: `Entry added successfully.` })
+    } catch (error) {
+        console.log({ message: 'Error writing to file:', details: error })
+    }
+}

src/utils/http.utils.ts

@@ -4,12 +4,11 @@ import { Express } from 'express';
 import { v4 as uuidv4 } from 'uuid'
 import { ConnectedHttpClient, ConnectedHttpServer } from "../transport/http";
 import { BehaviorSubject, Observable, Observer, Subject, Subscription } from "rxjs";
-import { ConnectionState, Transport, TransportEvent, TransportMessage } from '../interface/connector.interface';
-import { EventMessage, FisMessage } from '../interface/transport.interface';
 import { WrappedMessage } from './message.ordering';
 import axios, { AxiosError, AxiosResponse } from 'axios';
 import ConsoleLogger from './log.utils';
 import path from 'path';
+import { ConnectionState, FisMessage, GeneralEvent, TransportMessage } from '../interface/interface';
 const console: ConsoleLogger = new ConsoleLogger(`HttpUtils`, ['transport'])
 
 export function startHttpServer(port: number): Observable<ConnectedHttpClient> {
@@ -23,37 +22,35 @@ export function startHttpServer(port: number): Observable<ConnectedHttpClient> {
             console.log({ message: `Server running at http://localhost:${port}` });
         });
 
-        observer.next({
-            id: uuidv4(),
-            dateCreated: new Date(),
-            connectionState: new BehaviorSubject<ConnectionState>('ONLINE'),
-            instance: app
-        } as ConnectedHttpClient)
+
     })
 }
 
-export async function initiateClientToServer(url: string, event: Subject<TransportEvent>, connectedHttpServers: ConnectedHttpServer[], receiverProfileInfo: ConnectedHttpServer | undefined, browserEnv?: boolean,): Promise<ConnectedHttpServer> {
+export async function initiateClientToServer(transportServiceId: string, url: string, event: Subject<GeneralEvent<any>>, connectedHttpServers: ConnectedHttpServer[], receiverProfileInfo: ConnectedHttpServer | undefined, browserEnv?: boolean,): Promise<ConnectedHttpServer> {
     return new Promise((resolve, reject) => {
         if (browserEnv) {
             // logic here for using browser fetch
         } else { // axios methods
             if (receiverProfileInfo) {
                 console.log({ message: `Is Old profile, reconnecting with server` })
-                checkOwnClientInfo(receiverProfileInfo.id).then((profile: ConnectedHttpServer) => {
-                    receiverProfileInfo!.id = profile.id
+                checkOwnClientInfo(receiverProfileInfo.clientId).then((profile: ConnectedHttpServer) => {
+                    receiverProfileInfo!.clientId = profile.clientId
                     // console.log({ message: 'jsonfile', details: profile })
                     postAxiosRequest(url + 'profile', { name: 'Old Client', message: profile }).then((profileInfo: { name: string, message: { id: string } }) => {
                         console.log({ message: `Acknowledged as previous client. Id: ${profileInfo.message.id}` })
                         event.next({
                             id: uuidv4(),
+                            type: `Transport Event`,
                             event: 'Server Connected',
+                            date: new Date(),
                             data: {
                                 clientId: profileInfo.message.id,
                                 message: `Existing Http Channel ${profileInfo.message.id} re-established.`
-                            } as EventMessage
+                            },
+                            transport: `Http`
                         })
                         // Update Http instance record
-                        let clientObj: ConnectedHttpServer | undefined = connectedHttpServers.find(obj => obj.id === profileInfo.message.id)
+                        let clientObj: ConnectedHttpServer | undefined = connectedHttpServers.find(obj => obj.clientId === profileInfo.message.id)
                         console.log({ message: 'ClientObj', details: clientObj })
                         console.log({ message: 'ReceiverProfile', details: receiverProfileInfo })
                         if (clientObj) {
@@ -68,7 +65,7 @@ export async function initiateClientToServer(url: string, event: Subject<Transpo
                 }).catch((error) => {
                     console.error(error)
                     postAxiosRequest(url + 'profile', { name: 'New Client', data: null }).then((profileInfo: { name: string, message: any }) => {
-                        updateProfileAndPublishEvent((receiverProfileInfo as ConnectedHttpServer), profileInfo, event, connectedHttpServers).then((receiverProfileInfo) => {
+                        updateProfileAndPublishEvent(transportServiceId, (receiverProfileInfo as ConnectedHttpServer), profileInfo, event, connectedHttpServers).then((receiverProfileInfo) => {
                             resolve(receiverProfileInfo)
                         })
                     }).catch((error) => {
@@ -79,7 +76,7 @@ export async function initiateClientToServer(url: string, event: Subject<Transpo
             } else {
                 console.log({ message: `Is New profile, Connecting with server` })
                 postAxiosRequest(url + 'profile', { name: 'New Client', data: null }).then((profileInfo: { name: string, message: any }) => {
-                    updateProfileAndPublishEvent(receiverProfileInfo, profileInfo, event, connectedHttpServers).then((receiverProfileInfo) => {
+                    updateProfileAndPublishEvent(transportServiceId, receiverProfileInfo, profileInfo, event, connectedHttpServers).then((receiverProfileInfo) => {
                         resolve(receiverProfileInfo)
                     })
                 }).catch((error) => {
@@ -91,9 +88,9 @@ export async function initiateClientToServer(url: string, event: Subject<Transpo
 }
 
 // For client usage
-export function handleClientHttpConnection(url: string, server: ConnectedHttpServer): Observable<TransportEvent> {
-    return new Observable((eventNotification: Observer<TransportEvent>) => {
-        console.log({ message: `Long Poll Attempt for ${server.id}` })
+export function handleClientHttpConnection(url: string, server: ConnectedHttpServer): Observable<GeneralEvent<any>> {
+    return new Observable((eventNotification: Observer<GeneralEvent<any>>) => {
+        console.log({ message: `Long Poll Attempt for ${server.clientId}` })
         server.connectionState.next('ONLINE');
         let active: boolean = true; // Flag to control polling lifecycle
 
@@ -111,14 +108,17 @@ export function handleClientHttpConnection(url: string, server: ConnectedHttpSer
                         console.log({ message: 'Long Poll Response', details: data })
                         eventNotification.next({
                             id: uuidv4(),
+                            type: `Transport Event`,
                             event: 'New Message',
+                            date: new Date(),
                             data: {
                                 id: uuidv4(),
                                 dateCreated: new Date(),
-                                transport: Transport.Http,
-                                target: server.id,
+                                transport: `Http`,
+                                target: server.clientId,
                                 payload: data,
-                            } as TransportMessage,
+                            },
+                            transport: `Http`
                         });
                     } else if (response.status === 204) {
                         console.log({ message: 'No new messages from the server.' });
@@ -164,46 +164,54 @@ export function handleClientHttpConnection(url: string, server: ConnectedHttpSer
     });
 }
 
-function handleServerConnectionError(active: boolean, observer: Observer<TransportEvent>, server: ConnectedHttpServer): void {
+function handleServerConnectionError(active: boolean, observer: Observer<GeneralEvent<any>>, server: ConnectedHttpServer): void {
     server.connectionState.next('OFFLINE');
     console.log({ message: 'Server lost connection' });
     active = false; // Stop polling
     observer.next({
         id: uuidv4(),
+        type: `Transport Event`,
         event: 'Server Disconnected',
+        date: new Date(),
         data: {
-            clientId: server.id,
+            clientId: server.clientId,
             message: '',
             payload: {
-                time: new Date(),
                 objRef: server
             },
-        } as EventMessage,
+        },
+        transport: `Http`
     });
     observer.complete()
 }
 
-async function updateProfileAndPublishEvent(receiverProfileInfo: ConnectedHttpServer | undefined, profile: { name: string, message: any }, event: Subject<TransportEvent>, connectedHttpServers: ConnectedHttpServer[]): Promise<ConnectedHttpServer> {
+async function updateProfileAndPublishEvent(transportServiceId: string, receiverProfileInfo: ConnectedHttpServer | undefined, profile: { name: string, message: any }, event: Subject<GeneralEvent<any>>, connectedHttpServers: ConnectedHttpServer[]): Promise<ConnectedHttpServer> {
     return new Promise((resolve, reject) => {
-        console.log({ message: `Assigned client Name: ${(profile.message as ConnectedHttpServer).id}` })
+        console.log({ message: `Assigned client Name: ${(profile.message as ConnectedHttpServer).clientId}` })
         receiverProfileInfo = profile.message as ConnectedHttpServer
-        writeFile(profile.message as ConnectedHttpServer, (profile.message as ConnectedHttpServer).id).then(() => {
+        writeFile(profile.message as ConnectedHttpServer, (profile.message as ConnectedHttpServer).clientId).then(() => {
             event.next({
                 id: uuidv4(),
+                type: `Transport Event`,
                 event: `New Server`,
+                date: new Date(),
                 data: {
-                    clientId: (profile.message as ConnectedHttpServer).id,
-                    message: `New Http Channel ${(profile.message as ConnectedHttpServer).id} established.`
-                } as EventMessage
+                    clientId: (profile.message as ConnectedHttpServer).clientId,
+                    message: `New Http Channel ${(profile.message as ConnectedHttpServer).clientId} established.`
+                },
+                transport: `Http`
             })
             // broadcast event to allow retransmission to relase buffered messages
             event.next({
                 id: uuidv4(),
+                type: `Transport Event`,
                 event: `Server Connected`,
+                date: new Date(),
                 data: {
-                    clientId: (profile.message as ConnectedHttpServer).id,
-                    message: `Server ${(profile.message as ConnectedHttpServer).id} connected and ready to go.`
-                } as EventMessage
+                    clientId: (profile.message as ConnectedHttpServer).clientId,
+                    message: `Server ${(profile.message as ConnectedHttpServer).clientId} connected and ready to go.`
+                },
+                transport: `Http`
             })
         }).catch((error) => {
             reject(error)
@@ -211,9 +219,11 @@ async function updateProfileAndPublishEvent(receiverProfileInfo: ConnectedHttpSe
 
         // Update http instance record
         receiverProfileInfo = {
-            id: (profile.message as ConnectedHttpServer).id,
+            clientId: (profile.message as ConnectedHttpServer).clientId,
             dateCreated: new Date(),
-            connectionState: new BehaviorSubject<ConnectionState>(`ONLINE`)
+            connectionState: new BehaviorSubject<ConnectionState>(`ONLINE`),
+            transport: `Http`,
+            transportServiceId: transportServiceId
         }
         connectedHttpServers.push(receiverProfileInfo)
         resolve(receiverProfileInfo)
@@ -238,38 +248,43 @@ export async function postAxiosRequest(url: string, data: any): Promise<any> {
 }
 
 
-export function handleHttpClient(clientInfo: ConnectedHttpClient, connectedClientHttp: ConnectedHttpClient[]): Observable<TransportEvent> {
-    return new Observable((event: Observer<TransportEvent>) => {
+export function handleHttpClient(transportServiceId: string, clientInfo: ConnectedHttpClient, connectedClientHttp: ConnectedHttpClient[]): Observable<GeneralEvent<any>> {
+    return new Observable((event: Observer<GeneralEvent<any>>) => {
         clientInfo.instance.post('/profile', (req, res) => {
             // Client will declare this first before attempting to poll for response channel
-            handleProfile(clientInfo.instance, req.body, res, event, connectedClientHttp)
+            handleProfile(transportServiceId, clientInfo.instance, req.body, res, event, connectedClientHttp)
         });
     })
 }
 
-function handleProfile(app: Express, data: { name: `Old Client` | `New Client`, message: any }, res: Response, event: Observer<TransportEvent>, connectedClientHttp: ConnectedHttpClient[]): void {
+function handleProfile(transportServiceId: string, app: Express, data: { name: `Old Client` | `New Client`, message: any }, res: Response, event: Observer<GeneralEvent<any>>, connectedClientHttp: ConnectedHttpClient[]): void {
     if (data.name == `New Client`) {
         let clientInstance: ConnectedHttpClient = {
-            id: uuidv4(), // client should only be assigned at this level. And is passed around for reference pointing
+            clientId: uuidv4(), // client should only be assigned at this level. And is passed around for reference pointing
             dateCreated: new Date(),
             instance: app,
             connectionState: new BehaviorSubject<ConnectionState>(`OFFLINE`), // for now it's offline because it needs to establish the long polling first
-            responseStream: new Subject<WrappedMessage>()
+            responseStream: new Subject<WrappedMessage>(),
+            transportServiceId: transportServiceId,
+            transport: `Http`
         }
 
         // send to receiver for reference
         res.json({
-            name: `New Profile`, message: { id: clientInstance.id }
+            name: `New Profile`, message: { id: clientInstance.clientId }
         })
         // publish first event notification
         event.next({
             id: uuidv4(),
+            type: `Transport Event`,
             event: `New Client`,
+            date: new Date(),
             data: {
-                clientId: clientInstance.id,
-                message: `New Http Client Connected. Adapter ID assigned: ${clientInstance.id}`,
+                clientId: clientInstance.clientId,
+                message: `New Http Client Connected. Adapter ID assigned: ${clientInstance.clientId}`,
                 payload: clientInstance
-            } as EventMessage
+            },
+            transport: `Http`
         })
         // Update connected clientInstance info to adapter
         connectedClientHttp.push(clientInstance)
@@ -280,7 +295,7 @@ function handleProfile(app: Express, data: { name: `Old Client` | `New Client`,
         // update first
         let clientInstance: ConnectedHttpClient | undefined
         if (connectedClientHttp.length > 0) {
-            clientInstance = connectedClientHttp.find(obj => obj.id === data.message.id)
+            clientInstance = connectedClientHttp.find(obj => obj.clientId === data.message.id)
             handleFoundClient(clientInstance)
         } else {
             // for the case server itself got shit down or something
@@ -291,8 +306,8 @@ function handleProfile(app: Express, data: { name: `Old Client` | `New Client`,
         }
         function handleFoundClient(clientInstance: ConnectedHttpClient | undefined): void {
             if (clientInstance) {
-                console.log({ message: `Http Client ${clientInstance.id} Found` })
-                res.json({ name: 'Adjusted Profile', message: { id: clientInstance.id } })
+                console.log({ message: `Http Client ${clientInstance. clientId} Found` })
+                res.json({ name: 'Adjusted Profile', message: { id: clientInstance.clientId } })
                 // replace socket instance since the previous has been terminated
                 clientInstance.instance = app
                 // some explanation here. For the case where the server reads from the DB, no need to terminate subject, since all instances would be destroyed alongside the server shut down. This case is specificd only when there's a need to read from local file
@@ -303,12 +318,15 @@ function handleProfile(app: Express, data: { name: `Old Client` | `New Client`,
                 startListeningAndStreaming(app, clientInstance, event, true)
                 event.next({
                     id: uuidv4(),
+                    type: `Transport Event`,
                     event: 'Client Connected',
+                    date: new Date(),
                     data: {
-                        clientId: clientInstance.id,
-                        message: `Client ${clientInstance.id} connection re-established`,
+                        clientId: clientInstance.clientId,
+                        message: `Client ${clientInstance.clientId} connection re-established`,
                         payload: clientInstance
-                    } as EventMessage
+                    },
+                    transport: `Http`
                 })
 
             } else {
@@ -361,7 +379,7 @@ export function addClientToDB(entry: ConnectedHttpClient): void {
 
         // Append the new object to the array
         data.push({
-            id: entry.id,
+            id: entry.clientId,
             dateCreated: entry.dateCreated,
             connectionState: null,
             instance: null
@@ -376,19 +394,22 @@ export function addClientToDB(entry: ConnectedHttpClient): void {
 }
 
 // this is for server usage only
-export function startListeningAndStreaming(app: Express, client: ConnectedHttpClient, eventListener: Observer<TransportEvent>, oldClient?: boolean): void {
+export function startListeningAndStreaming(app: Express, client: ConnectedHttpClient, eventListener: Observer<GeneralEvent<any>>, oldClient?: boolean): void {
     /* Generally, we don't need this unless in the case of being the receiver */
     app.post('/message', (req, res) => {
         eventListener.next({
             id: uuidv4(),
+            type: `Transport Event`,
             event: 'New Message',
+            date: new Date(),
             data: {
                 id: uuidv4(),
                 dateCreated: new Date(),
-                transport: Transport.Http,
-                target: client.id, // this ref to be associated with the client/channel
+                transport: `Http`,
+                target: client.clientId, // this ref to be associated with the client/channel
                 payload: req.body
-            } as TransportMessage
+            } as TransportMessage,
+            transport: `Http`
         })
         res.json(`Received ${((req.body as WrappedMessage)?.payload as FisMessage)?.header?.messageID ?? `Undefined`}`)
     })
@@ -400,12 +421,15 @@ export function startListeningAndStreaming(app: Express, client: ConnectedHttpCl
         // notify it's associated retransmission to start releaseing buffer
         eventListener.next({
             id: uuidv4(),
+            type: `Transport Event`,
             event: oldClient ? 'Client Re-connected' : `Client Connected`,
+            date: new Date(),
             data: {
-                clientId: client.id,
-                message: `Socket Client ${oldClient ? `Re-Connected` : `Connected`}. Adapter ID assigned: ${client.id}`,
+                clientId: client.clientId,
+                message: `Socket Client ${oldClient ? `Re-Connected` : `Connected`}. Adapter ID assigned: ${client.clientId}`,
                 payload: client
-            } as EventMessage
+            },
+            transport: `Http`
         })
 
         // Flag to track if the response has been sent
@@ -415,7 +439,7 @@ export function startListeningAndStreaming(app: Express, client: ConnectedHttpCl
         const subscription = client.responseStream.asObservable().subscribe({
             next: (message: WrappedMessage) => {
                 if (!responseSent) {
-                    console.log({ message: `Sending data ${message.thisMessageID} to client ${client.id}}` });
+                    console.log({ message: `Sending data ${message.thisMessageID} to client ${client.clientId}}` });
                     res.json(message); // Send the data to the client
                     responseSent = true; // Mark response as sent
                     subscription.unsubscribe(); // Unsubscribe to close this request
@@ -452,17 +476,17 @@ export function startListeningAndStreaming(app: Express, client: ConnectedHttpCl
         // Handle client disconnection
         res.on('close', () => {
             if (!responseSent) {
-                console.error({ message: `Http Client ${client.id} disconnected` });
+                console.error({ message: `Http Client ${client.clientId} disconnected` });
                 eventListener.next({
                     id: uuidv4(),
+                    type: `Transport Event`,
                     event: 'Client Disconnected',
+                    date: new Date(),
                     data: {
-                        clientId: client.id,
-                        payload: {
-                            time: new Date()
-                        }
-                    } as EventMessage
-                } as TransportEvent)
+                        clientId: client.clientId,
+                    },
+                    transport: `Http`
+                })
                 client.connectionState.next(`OFFLINE`)
                 subscription.unsubscribe(); // Ensure cleanup
             }

src/utils/socket.utils.ts

@@ -4,11 +4,11 @@ import { Server, Socket as SocketForConnectedClient } from 'socket.io';
 import { io, Socket as SocketForConnectedServer } from 'socket.io-client';
 import * as fs from 'fs'
 import { v4 as uuidv4 } from 'uuid'
-import { ConnectionState, Transport, TransportEvent, TransportMessage } from '../interface/connector.interface';
 import { ConnectedSocketClient, ConnectedSocketServer } from '../transport/websocket';
-import { EventMessage } from '../interface/transport.interface';
 import ConsoleLogger from './log.utils';
 import path from 'path';
+import { ConnectionState, GeneralEvent, TransportMessage } from '../interface/interface';
+import { addClientToDB, checkIfClientExists, checkOwnClientInfo, writeFile } from './general.utils';
 const console: ConsoleLogger = new ConsoleLogger(`SocketUtils`, ['transport'])
 
 export function startSocketServer(port: number): Observable<SocketForConnectedClient> {
@@ -60,16 +60,16 @@ export async function startClientSocketConnection(serverUrl: string): Promise<So
 }
 
 // After establishing connection to the server, set up the credentials, confirm whether or not if there's any credentials, if not ask for one from the server
-export function handleClientSocketConnection(socket: SocketForConnectedServer, serversConnected: ConnectedSocketServer[]): Observable<TransportEvent> {
-    return new Observable((eventNotification: Observer<TransportEvent>) => {
+export function handleClientSocketConnection(transportServiceId: string, socket: SocketForConnectedServer, serversConnected: ConnectedSocketServer[]): Observable<GeneralEvent<any>> {
+    return new Observable((eventNotification: Observer<GeneralEvent<any>>) => {
         let buffer: any[] = []
         let receiverProfileInfo!: ConnectedSocketServer
 
         // Listen for a connection event
         socket.on('connect', () => {
             console.log({ message: `Connected to the server ${socket.id} ` })
-            if (receiverProfileInfo?.id) {
-                checkOwnClientInfo(receiverProfileInfo.id).then((profile: { id: string }) => {
+            if (receiverProfileInfo?.clientId) {
+                checkOwnClientInfo(receiverProfileInfo.clientId).then((profile: { id: string }) => {
                     socket.emit('profile', {
                         name: 'Old Client',
                         data: profile
@@ -81,7 +81,7 @@ export function handleClientSocketConnection(socket: SocketForConnectedServer, s
                     })
                 })
             } else {
-                socket.emit('profile', {
+                socket.emit('profile', { 
                     name: 'New Client',
                     data: null
                 })
@@ -95,12 +95,14 @@ export function handleClientSocketConnection(socket: SocketForConnectedServer, s
                 // publish to event
                 eventNotification.next({
                     id: uuidv4(),
+                    type: `Transport Event`,
                     event: 'New Message',
+                    date: new Date(),
                     data: {
                         id: uuidv4(),
                         dateCreated: new Date(),
-                        transport: Transport.Websocket,
-                        target: receiverProfileInfo.id,
+                        transport: `Websocket`,
+                        target: receiverProfileInfo.clientId,
                         payload: msg
                     } as TransportMessage
                 })
@@ -116,51 +118,62 @@ export function handleClientSocketConnection(socket: SocketForConnectedServer, s
                 console.log({ message: `Assigned client Name: ${data.message.id}` })
                 // Update websocket instance record
                 receiverProfileInfo = {
-                    id: data.message.id,
+                    clientId: data.message.id,
                     dateCreated: new Date(),
                     socketInstance: socket,
-                    connectionState: new BehaviorSubject<ConnectionState>(`ONLINE`)
+                    connectionState: new BehaviorSubject<ConnectionState>(`ONLINE`),
+                    transport: 'Websocket',
+                    transportServiceId: transportServiceId
                 }
-                writeFile(data.message as ConnectedSocketServer, (data.message as ConnectedSocketServer).id).then(() => {
+                writeFile(data.message as ConnectedSocketServer, (data.message as ConnectedSocketServer).clientId).then(() => {
                     // broadcast event to allow transmission manager to instantiate transmission components
                     eventNotification.next({
                         id: uuidv4(),
+                        type: 'Transport Event',
                         event: `New Server`,
+                        date: new Date(),
                         data: {
-                            clientId: (data.message as ConnectedSocketServer).id,
-                            message: `New Websocket Channel ${(data.message as ConnectedSocketServer).id} established.`
-                        } as EventMessage
+                            clientId: (data.message as ConnectedSocketServer).clientId,
+                            message: `New Websocket Channel ${(data.message as ConnectedSocketServer).clientId} established.`
+                        },
+                        transport: 'Websocket'
                     })
                     // broadcast event to allow retransmission to relase buffered messages
                     eventNotification.next({
                         id: uuidv4(),
+                        type: 'Transport Event',
                         event: `Server Connected`,
+                        date: new Date(),
                         data: {
-                            clientId: (data.message as ConnectedSocketServer).id,
-                            message: `Server ${(data.message as ConnectedSocketServer).id} connected and ready to go.`
-                        } as EventMessage
+                            clientId: (data.message as ConnectedSocketServer).clientId,
+                            message: `Server ${(data.message as ConnectedSocketServer).clientId} connected and ready to go.`
+                        },
+                        transport: 'Websocket'
                     })
                 }).catch((error) => { }) // do nothing at the moment. 
                 serversConnected.push(receiverProfileInfo)
             }
             if (data.name == 'Adjusted Profile') {
-                console.log({ message: `Adjusted client Name: ${(data.message as ConnectedSocketServer).id}` })
+                console.log({ message: `Adjusted client Name: ${(data.message as ConnectedSocketServer).clientId}` })
                 // Update websocket instance record
-                let clientObj: ConnectedSocketServer | undefined = serversConnected.find(obj => obj.id === data.message.id)
+                let clientObj: ConnectedSocketServer | undefined = serversConnected.find(obj => obj.clientId === data.message.id)
                 if (clientObj) {
                     clientObj.socketInstance = socket
                     clientObj.connectionState.next('ONLINE')
                     console.log({
-                        message: `Just to make sure they are pointed accurately: This should be ONLINE: ${receiverProfileInfo.connectionState.getValue()} !! Id match? ${receiverProfileInfo.id == clientObj.id ? true : false} && compare ${clientObj.id}`,
+                        message: `Just to make sure they are pointed accurately: This should be ONLINE: ${receiverProfileInfo.connectionState.getValue()} !! Id match? ${receiverProfileInfo.clientId == clientObj.clientId ? true : false} && compare ${clientObj.clientId}`,
                     })
                     // broadcast event to allow retransmission to release buffer
                     eventNotification.next({
                         id: uuidv4(),
+                        type: `Transport Event`,
                         event: 'Server Connected',
+                        date: new Date(),
                         data: {
-                            clientId: (data.message as ConnectedSocketServer).id,
-                            message: `Existing Websocket Channel ${(data.message as ConnectedSocketServer).id} re-established.`
-                        } as EventMessage
+                            clientId: (data.message as ConnectedSocketServer).clientId,
+                            message: `Existing Websocket Channel ${(data.message as ConnectedSocketServer).clientId} re-established.`
+                        },
+                        transport: 'Websocket'
                     })
                 }
             }
@@ -178,15 +191,18 @@ export function handleClientSocketConnection(socket: SocketForConnectedServer, s
 
         // Handle disconnection
         socket.on('disconnect', () => {
-            console.log({ message: `Socket Server ${receiverProfileInfo.id} Disconnected` })
+            console.log({ message: `Socket Server ${receiverProfileInfo.clientId} Disconnected` })
             if (receiverProfileInfo) {
                 eventNotification.next({
                     id: uuidv4(),
+                    type: `Transport Event`,
                     event: `Server Disconnected`,
+                    date: new Date(),
                     data: {
-                        clientId: receiverProfileInfo.id,
-                        message: `Server for Channel ${receiverProfileInfo.id} disconnected.`
-                    } as EventMessage
+                        clientId: receiverProfileInfo.clientId,
+                        message: `Server for Channel ${receiverProfileInfo.clientId} disconnected.`
+                    },
+                    transport: `Websocket`
                 })
                 receiverProfileInfo.connectionState.next(`OFFLINE`)
             }
@@ -195,32 +211,37 @@ export function handleClientSocketConnection(socket: SocketForConnectedServer, s
 }
 
 // For SERVER Usage: set up socket listeners to start listening for different events
-export function handleNewSocketClient(socket: SocketForConnectedClient, connectedClientSocket: ConnectedSocketClient[]): Observable<TransportEvent> {
-    return new Observable((event: Observer<TransportEvent>) => {
+export function handleNewSocketClient(transportServiceId: string, socket: SocketForConnectedClient, connectedClientSocket: ConnectedSocketClient[]): Observable<GeneralEvent<any>> {
+    return new Observable((event: Observer<GeneralEvent<any>>) => {
         console.log({ message: `Setting up listeners for socket:${socket.id}` })
         // returns the socket client instance 
         // listen to receiver's initiotion first before assigning 'credentials'
         socket.on(`profile`, (message: { name: string, data: any }) => {
             if (message.name == 'New Client') {
                 let clientInstance: ConnectedSocketClient = {
-                    id: uuidv4(), // client should only be assigned at this level. And is passed around for reference pointing
+                    clientId: uuidv4(), // client should only be assigned at this level. And is passed around for reference pointing
                     dateCreated: new Date(),
                     socketInstance: socket,
-                    connectionState: new BehaviorSubject<ConnectionState>(`OFFLINE`)
+                    connectionState: new BehaviorSubject<ConnectionState>(`OFFLINE`),
+                    transport: 'Websocket',
+                    transportServiceId: transportServiceId
                 }
                 // send to receiver for reference
                 socket.emit('profile', {
-                    name: `New Profile`, message: { id: clientInstance.id }
+                    name: `New Profile`, message: { id: clientInstance.clientId }
                 })
                 // publish first event notification
                 event.next({
                     id: uuidv4(),
+                    type: 'Transport Event',
                     event: `New Client`,
+                    date: new Date(),
                     data: {
-                        clientId: clientInstance.id,
-                        message: `New Socket Client Connected. Adapter ID assigned: ${clientInstance.id}`,
+                        clientId: clientInstance.clientId,
+                        message: `New Socket Client Connected. Adapter ID assigned: ${clientInstance.clientId}`,
                         payload: clientInstance
-                    } as EventMessage
+                    },
+                    transport: 'Websocket'
                 })
                 // Update connected clientInstance info to adapter
                 connectedClientSocket.push(clientInstance)
@@ -230,7 +251,7 @@ export function handleNewSocketClient(socket: SocketForConnectedClient, connecte
                 // update first
                 let clientInstance: ConnectedSocketClient | undefined
                 if (connectedClientSocket.length > 0) {
-                    clientInstance = connectedClientSocket.find(obj => obj.id === message.data.id)
+                    clientInstance = connectedClientSocket.find(obj => obj.clientId === message.data.id)
                     handleFoundClient(clientInstance)
                 } else {
                     // for the case server itself got shit down or something
@@ -243,13 +264,12 @@ export function handleNewSocketClient(socket: SocketForConnectedClient, connecte
                 }
                 function handleFoundClient(clientInstance: ConnectedSocketClient | undefined) {
                     if (clientInstance) {
-                        console.log({ message: `Socket Client ${clientInstance.id} Found` })
-                        socket.emit('profile', { name: 'Adjusted Profile', message: { id: clientInstance.id } })
+                        console.log({ message: `Socket Client ${clientInstance.clientId} Found` })
+                        socket.emit('profile', { name: 'Adjusted Profile', message: { id: clientInstance.clientId } })
                         // replace socket instance since the previous has been terminated
                         clientInstance.socketInstance = socket
                         // need to start listening again, because it's assigned a different socket instance this time round
                         startListening(socket, clientInstance, event, true)
-
                     } else {
                         console.log({ message: `Profile Not Found` })
                         socket.emit('profile', { name: 'Error', message: 'Receiver Profile Not found' })
@@ -260,133 +280,21 @@ export function handleNewSocketClient(socket: SocketForConnectedClient, connecte
     })
 }
 
-async function writeFile(data: ConnectedSocketServer, filename: string): Promise<boolean> {
-    return new Promise((resolve, reject) => {
-        // Ensure the folder exists
-        const folderPath = process.env.FolderPath as string
-        console.log({ message: folderPath })
-        // const folderPath = path.join(__dirname, folder);
-        if (!fs.existsSync(folderPath)) {
-            fs.mkdirSync(folderPath, { recursive: true }); // Create folder if it doesn't exist
-        } else {
-            console.log({ message: 'Folder already exist' })
-        }
-
-        // Construct the full file path (include the folder)
-        const filePath = path.join(folderPath, `${filename}.json`);
-
-        // Write JSON data to a file
-        fs.writeFile(filePath, JSON.stringify(data, null, 2), (err) => {
-            if (err) {
-                console.log({ message: 'Error writing file', details: err });
-                reject(false);
-            } else {
-                console.log({ message: 'File has been written', details: filePath });
-                resolve(true);
-            }
-        });
-    });
-}
-
-/* For Internal Usage only. Temporary serve as a way for server to keep track of clients. To be replaced in the future with better alternatives. */
-export function addClientToDB(entry: ConnectedSocketClient): void {
-    try {
-        let data: ConnectedSocketClient[] = [];
-        let filePath = process.env.FolderPath as string + 'clients.json'
-        // Check if the file exists and load existing data
-        if (fs.existsSync(filePath)) {
-            const fileContent = fs.readFileSync(filePath, 'utf-8');
-            data = JSON.parse(fileContent);
-        }
-
-        // Append the new details to the array
-        data.push({
-            id: entry.id,
-            dateCreated: entry.dateCreated,
-            connectionState: null,
-            socketInstance: null
-        } as unknown as ConnectedSocketClient);
-
-        // Write the updated array back to the file
-        fs.writeFileSync(filePath, JSON.stringify(data, null, 2), 'utf-8');
-        console.log({ message: `Entry added successfully.` })
-    } catch (error) {
-        console.log({ message: 'Error writing to file:', details: error })
-    }
-}
-
-export async function checkIfClientExists(id: string): Promise<ConnectedSocketClient> {
-    return new Promise((resolve, reject) => {
-        try {
-            // Check if the file exists
-            let filePath = process.env.FolderPath as string + 'clients.json'
-            if (!fs.existsSync(filePath)) {
-                console.log({ message: `File does not exist.` })
-                reject('File does not exist');
-            }
-
-            // Read and parse the data
-            const fileContent = fs.readFileSync(filePath, 'utf-8');
-            const data: any[] = JSON.parse(fileContent);
-
-            // Check if an details with the given id exists
-            let obj = data.find(entry => entry.id === id);
-
-            if (obj) {
-                console.log({ message: "Client with ID ${id} exists." })
-            } else {
-                console.log({ message: `Client with ID ${id} does not exist.` })
-            }
-            resolve(obj);
-        } catch (error) {
-            reject(`Error reading the file`)
-        }
-    })
-}
-
-
-// Check if filename exists. Return profile information if there's any
-export async function checkOwnClientInfo(filename?: string): Promise<{ id: string }> {
-    return new Promise((resolve, reject) => {
-        // Check if the file exists
-        let filePath = process.env.FolderPath as string
-        if (fs.existsSync(`${filePath}${filename}.json`)) {
-            try {
-                // Read the file contents
-                const fileData = fs.readFileSync(filePath + `${filename}.json`, 'utf8');
-
-                // If the file is empty, return an error
-                if (fileData.trim() === "") {
-                    throw new Error("File is empty");
-                }
-
-                // Parse and return the data if present
-                const jsonData = JSON.parse(fileData);
-                resolve(jsonData)
-
-            } catch (err) {
-                // Handle parsing errors or other file-related errors
-                console.log({ message: "Error reading or parsing file:", details: err })
-                reject('');
-            }
-        } else {
-            console.log({ message: `File does not exist. Id: ${filename}` })
-            reject('');
-        }
-    })
-}
 
 // this is for server usage only
-export function startListening(socket: SocketForConnectedClient, client: ConnectedSocketClient, eventListener: Observer<TransportEvent>, oldClient?: boolean): void {
+export function startListening(socket: SocketForConnectedClient, client: ConnectedSocketClient, eventListener: Observer<GeneralEvent<any>>, oldClient?: boolean): void {
     // notify it's associated retransmission to start releaseing buffer
     eventListener.next({
         id: uuidv4(),
+        type: 'Transport Event',
         event: oldClient ? 'Client Re-connected' : `Client Connected`,
+        date: new Date(),
         data: {
-            clientId: client.id,
-            message: `Socket Client ${oldClient ? `Re-Connected` : `Connected`}. Adapter ID assigned: ${client.id}`,
-            payload: client
-        } as EventMessage
+            clientId: client.clientId,
+            message: `Socket Client ${oldClient ? `Re-Connected` : `Connected`}. Adapter ID assigned: ${client.clientId}`,
+            payloclientId: client
+        },
+        transport: 'Websocket'
     })
     // Resume operation
     // some explanation here. For the case where the server reads from the DB, no need to terminate subject, since all instances would be destroyed alongside the server shut down. This case is specificd only when there's a need to read from local file
@@ -398,15 +306,17 @@ export function startListening(socket: SocketForConnectedClient, client: Connect
 
     /* Generally, we don't need this unless in the case of being the receiver */
     socket.on('message', (message: any) => {
-        console.log({ message: `Message from client ${client.id}`, details: message })
+        console.log({ message: `Message from client ${client.clientId}`, details: message })
         eventListener.next({
             id: uuidv4(),
+            type: 'Transport Event',
             event: 'New Message',
+            date: new Date(),
             data: {
                 id: uuidv4(),
                 dateCreated: new Date(),
-                transport: Transport.Websocket,
-                target: client.id, // this ref to be associated with the client/channel
+                transport: `Websocket`,
+                target: client.clientId, // this ref to be associated with the client/channel
                 payload: message
             } as TransportMessage
         })
@@ -415,16 +325,19 @@ export function startListening(socket: SocketForConnectedClient, client: Connect
     socket.on('disconnect', () => {
         eventListener.next({
             id: uuidv4(),
+            type: `Transport Event`,
             event: 'Client Disconnected',
+            date: new Date(),
             data: {
-                clientId: client.id,
+                clientId: client.clientId,
                 message: '',
                 payload: {
                     time: new Date()
                 }
-            } as EventMessage
+            },
+            transport: 'Websocket'
         })
-        eventListener.error(`Client ${client.id} disconnected. Terminating this observable event for this client socket...`)
+        eventListener.error(`Client ${client.clientId} disconnected. Terminating this observable event for this client socket...`)
         eventListener.complete()
     })
 }