implemented and tested with basic subscribable and interface update

src/adapters/adapter.manager.ts

@@ -3,7 +3,7 @@ How it works: It will first connect and listen to any underlying transport proto
 It then listens to the event there, particularly if a new server clients has been connected. It then creates the adapters based on that client and stores them in memory. It will also
 broadcast the aforementioned event, so that all those interested in the instantiated of a new adapter event can also has the reference for that adapter. For example, transmitter components
 wants to subscribe for transmitter adapter, so this manager will find through it's adapter's record and publish it to the transmitter, whilst also publish relevant event regarding the 
-instantiation of new adapters, so that the aforementioned transmitter components will have access to the adapter, and can begin it's transmission process.*/
+instantiation of new adapters, so that the aforementioned transmitter components will have access to the adapter, and can begin it's transmission process. */
 import { filter, map, Observable, Observer, Subject, Subscription } from "rxjs"
 import { v4 as uuidv4 } from 'uuid'
 import ConsoleLogger from "../utils/log.utils"
@@ -25,21 +25,19 @@ export class AdapterManager extends AdapterManagerBase {
         this.initializeAdapterInstantiation(generalEvent, this.event)
     }
 
-    public subscribeForAdapters(clientId: string, role: TransmissionRole): Observable<AdapterInterface> {
-        return new Observable((adapters: Observer<AdapterInterface>) => {
-            this.adapters.forEach((adapter: AdapterInterface) => {
-                if (adapter.getAdapterProfile(`clientId`) === clientId && adapter.getAdapterProfile(`role`) === role) {
-                    adapters.next(adapter)
-                }
+    // this will be called by Transmission Manager
+    public subscribeForAdapters(): Observable<AdapterInterface<any>> {
+        return new Observable((adapters: Observer<AdapterInterface<any>>) => {
+            // this one is just to pass the already instantiated adapters
+            this.adapters.forEach((adapter: AdapterInterface<any>) => {
+                adapters.next(adapter)
             })
 
             // According to chatgpt, any new adpaters that I insert into the array therafter will not be picked up, so need to manually listen to adapter evnet
             const subscription: Subscription = this.event.pipe(
                 filter(event => event.type === `Adapter Event`),
                 filter(event => event.event === `New Adapter`),
-                filter(event => (event.data as AdapterInterface).getAdapterProfile(`role`) === role),
-                filter(event => (event.data as AdapterInterface).getAdapterProfile(`clientId`) === clientId),
-                map(event => { return (event.data as AdapterInterface) })
+                map(event => { return (event.data as AdapterInterface<any>) })
             ).subscribe(adapter => {
                 adapters.next(adapter)
             })
@@ -60,7 +58,7 @@ export class AdapterManager extends AdapterManagerBase {
             filter(event => event.type === `Transport Event`),
         ).subscribe(event => {
             if (event.event == `New Client` || event.event == `New Server`) {
-                let adapters: AdapterInterface[] = this.instantiateAdapterComponents(event.data as ClientObject, event.transport, event.transportServiceId as string)
+                let adapters: AdapterInterface<any>[] = this.instantiateAdapterComponents(event.data as ClientObject, event.transport, event.transportServiceId as string)
                 adapters.forEach(adapter => {
                     transportEvent.next({
                         id: uuidv4(),
@@ -70,7 +68,7 @@ export class AdapterManager extends AdapterManagerBase {
                         data: adapter,
                         transport: event.transport,
                         transportServiceId: event.transportServiceId
-                    } as GeneralEvent<AdapterInterface>)
+                    } as GeneralEvent<AdapterInterface<any>>)
                 })
             } else if (event.event == `Client Disconnected` || event.event == `Server Disconnected`) {
                 // logic here ... pass...
@@ -110,13 +108,13 @@ export class AdapterManager extends AdapterManagerBase {
     }
     /* At the moment, all adapters are instantiated based on which transport the client is connected. For multi type adapters instantiation logic for one client, this 
     will be for future enhancements. */
-    private instantiateAdapterComponents(clientRef: ClientObject, transportType: TransportType, transportServiceId: string): AdapterInterface[] {
+    private instantiateAdapterComponents(clientRef: ClientObject, transportType: TransportType, transportServiceId: string): AdapterInterface<any>[] {
         let transportService = this.transportServiceArray.find(obj => obj.getInfo().transportServiceId === transportServiceId)
         if (transportService) {
             this.console.log({ message: `Instantiating ${transportType} adapters for client ${clientRef.clientId}` });
-            const adapters: AdapterInterface[] = [
-                new TransmitterAdapter(clientRef.clientId, transportService.getInfo().transport, transportService, `Transmitter`),
-                new ReceiverAdapter(clientRef.clientId, transportService.getInfo().transport, transportService, `Receiver`),
+            let adapters: AdapterInterface<any>[] = [
+                new TransmitterAdapter<any>(clientRef.clientId, transportService.getInfo().transport, transportService, `Transmitter`),
+                new ReceiverAdapter<any>(clientRef.clientId, transportService.getInfo().transport, transportService, `Receiver`),
             ];
             this.adapters.concat(adapters)
             return adapters;

src/adapters/adapter.receiver.ts

@@ -8,7 +8,7 @@ import { FisMessage, GeneralEvent, TransportType, TransportMessage, TransportSer
 
 dotenv.config();
 
-export class ReceiverAdapter extends AdapterBase {
+export class ReceiverAdapter<T> extends AdapterBase<T> {
 
     constructor(clientId: string, adapterType: TransportType, transportService: TransportServiceInterface, role: TransmissionRole) {
         super(clientId, adapterType, transportService, role)

src/adapters/adapter.transmitter.ts

@@ -6,7 +6,7 @@ import { AdapterBase } from '../base/adapter.base';
 import { ClientObject, ConnectionState, FisMessage, TransportType, TransportMessage, TransportServiceInterface, TransmissionRole } from '../interface/interface';
 
 dotenv.config();
-export class TransmitterAdapter extends AdapterBase {
+export class TransmitterAdapter<T> extends AdapterBase<T> {
 
     constructor(clientId: string, adapterType: TransportType, transportService: TransportServiceInterface, role: TransmissionRole) {
         super(clientId, adapterType, transportService, role)

src/base/adapter.base.ts

@@ -1,11 +1,11 @@
-import { BehaviorSubject, distinctUntilChanged, filter, map, Observable, Subject } from "rxjs";
+import { BehaviorSubject, distinctUntilChanged, filter, map, Observable, Observer, Subject, Unsubscribable } from "rxjs";
 import { AdapterInterface, ConnectionState, TransmissionRole, TransportType, TransportServiceInterface, AdapterProfile, ClientObject } from "../interface/interface";
 import ConsoleLogger from "../utils/log.utils";
 import { v4 as uuidv4 } from 'uuid'
 
 /* This transport manager will be instantiating the necessary transport to deal with tranmission and receiving from different receivers
 So how?: */
-export class AdapterBase implements AdapterInterface {
+export class AdapterBase<T> implements AdapterInterface<T> {
     protected console!: ConsoleLogger
     protected adapterProfile!: AdapterProfile
 
@@ -15,6 +15,10 @@ export class AdapterBase implements AdapterInterface {
         this.setupConnectionState(this.adapterProfile.transportService)
     }
 
+    subscribe(observer: Observer<T>): Unsubscribable {
+        throw new Error("Method not implemented.");
+    }
+
     protected setAdapterProfile(clientId: string, adapterType: TransportType, transportService: TransportServiceInterface, role: TransmissionRole): void {
         this.adapterProfile = {} as AdapterProfile
         this.adapterProfile.id = uuidv4()
@@ -25,25 +29,22 @@ export class AdapterBase implements AdapterInterface {
         this.adapterProfile.transportService = transportService
     }
 
-    getAdapterProfile(type?: `id` | `clientId` | `role` | `transportId` | `transportType` | `connectionState`): string | Observable<ConnectionState> | AdapterProfile | undefined {
-        if (!type) {
-            return this.adapterProfile
-        } else if (type == `id`) {
-            return this.adapterProfile.id
-        } else if (type == `clientId`) {
-            return this.adapterProfile.clientId
-        } else if (type == `connectionState`) {
-            return this.adapterProfile.connectionState.asObservable()
-        } else if (type == `role`) {
-            return this.adapterProfile.role
-        } else if (type == `transportType`) {
-            return this.adapterProfile.transportType
-        } else if (type == `transportId`) {
-            return this.adapterProfile.transportService.getInfo().transportServiceId
-        }
+    public getAdapterProfile(type?: `id` | `clientId` | `role` | `transportId` | `transportType` | `connectionState`): string | Observable<ConnectionState> | AdapterProfile | undefined {
+        if (!type) return this.adapterProfile;
 
+        const lookup: Record<string, any> = {
+            id: this.adapterProfile.id,
+            clientId: this.adapterProfile.clientId,
+            role: this.adapterProfile.role,
+            transportType: this.adapterProfile.transportType,
+            transportId: this.adapterProfile.transportService.getInfo().transportServiceId,
+            connectionState: this.adapterProfile.connectionState.asObservable()
+        };
+
+        return lookup[type];
     }
 
+
     protected setupConnectionState(transportService: TransportServiceInterface): void {
         transportService.subscribeForEvent().pipe(
             filter(event => event.type === `Transport Event`),

src/base/adapter.manager.base.ts

@@ -7,13 +7,13 @@ export class AdapterManagerBase implements AdapterManagerInterface {
     protected browserEnv!: boolean
     protected event!: Subject<GeneralEvent<any>>
     protected transportServiceArray: TransportServiceInterface[] = []
-    protected adapters: AdapterInterface[] = []
+    protected adapters: AdapterInterface<any>[] = []
 
     constructor() {
         // logic here
     }
 
-    subscribeForAdapters(receiverId: string, role: TransmissionRole): Observable<AdapterInterface> {
+    subscribeForAdapters(): Observable<AdapterInterface<any>> {
         throw new Error("Method not implemented.")
     }
 

src/base/msg.transmission.base.ts

@@ -1,11 +1,27 @@
-import { Observable, Subject, Subscription } from 'rxjs';
-import { AdapterInterface, GeneralEvent, MessageTransmissionInterface, TransmissionProfile } from '../interface/interface'
+import { Observable, Observer, Subject, Subscription, Unsubscribable } from 'rxjs';
+import { AdapterInterface, GeneralEvent, MessageTransmissionInterface, TransmissionConfig, TransmissionProfile, TransmissionRole } from '../interface/interface'
+import { v4 as uuidv4 } from 'uuid'
+export class MessageTransmissionBase<T> implements MessageTransmissionInterface<T> {
+    protected transmissionProfile!: TransmissionProfile
+    protected adapters: AdapterInterface<any>[] = []
 
-export class MessageTransmissionBase implements MessageTransmissionInterface {
-    protected profile!: TransmissionProfile
-    protected adapters: AdapterInterface[] = []
-    constructor() {
+    constructor(profile: TransmissionConfig, event: Observable<GeneralEvent<T>>) {
         // logic here
     }
+
+    public getTransmissionProfile(type?: `config` | `id` | `role`): TransmissionProfile | string | TransmissionConfig | undefined {
+        return type ? this.transmissionProfile[type] : this.transmissionProfile;
+    }
+
+    public subscribe(observer: Observer<T> | Observable<T>): Unsubscribable | null {
+        throw new Error('Method not implemented.');
+    }
+
+    protected setTransmissionProfile(role: TransmissionRole, config: TransmissionConfig): void {
+        this.transmissionProfile = {} as TransmissionProfile
+        this.transmissionProfile.id = uuidv4()
+        this.transmissionProfile.config = config
+        this.transmissionProfile.role = role
+    }
 }
 

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

@@ -1,29 +1,34 @@
 
 import { Subject } from 'rxjs';
-import { GeneralEvent, MessageReceiverInterface, MessageRequestResponseInterface, MessageTransmissionInterface, MessageTransmissionManagerInterface, MessageTransmitterInterface, TransmissionProfile } from '../interface/interface';
+import { AdapterInterface, GeneralEvent, MessageReceiverInterface, MessageRequestResponseInterface, MessageTransmissionInterface, MessageTransmissionManagerInterface, MessageTransmitterInterface, TransmissionConfig, TransmissionProfile, TransmissionRole } from '../interface/interface';
 import { AdapterManager } from '../adapters/adapter.manager';
 import ConsoleLogger from '../utils/log.utils';
 
-export class MessageTransmissionManagerBase implements MessageTransmissionManagerInterface {
+export class MessageTransmissionManagerBase<T> implements MessageTransmissionManagerInterface<T> {
     protected console!: ConsoleLogger
     protected browserEnv!: boolean
-    protected transmissionRef: MessageTransmissionInterface[] = []
+    protected transmissionObjects: MessageTransmissionInterface<T>[] = []
     protected adapterManager!: AdapterManager
-    protected event!: Subject<GeneralEvent<any>>
+    protected globalEvent!: Subject<GeneralEvent<any>>
+    protected transmissionEvent!: Subject<GeneralEvent<any>>
 
     constructor() {
         // logic here
     }
-    
-    getTransmitter(profile: TransmissionProfile): MessageTransmitterInterface {
+
+    getTransmissionObject(config: TransmissionConfig, role: TransmissionRole): MessageTransmissionInterface<T> | null {
+        throw new Error(`Method not implemented`)
+    }
+
+    getTransmitter(config: TransmissionConfig): MessageTransmitterInterface<T> {
         throw new Error(`Method not implemented`)
     }
 
-    getReceiver(profile: TransmissionProfile): MessageReceiverInterface {
+    getReceiver(config: TransmissionConfig): MessageReceiverInterface<T> {
         throw new Error(`Method not implemented`)
     }
 
-    getRequestResponse(profile: TransmissionProfile, transmitterInstance: MessageTransmitterInterface, receiverInstance: MessageReceiverInterface): MessageRequestResponseInterface {
+    getRequestResponse(config: TransmissionConfig, transmitterInstance: MessageTransmitterInterface<T>, receiverInstance: MessageReceiverInterface<T>): MessageRequestResponseInterface<T> {
         throw new Error(`Method not implemented`)
     }
 

src/interface/interface.ts

@@ -1,4 +1,4 @@
-import { BehaviorSubject, Observable } from "rxjs"
+import { BehaviorSubject, Observable, Observer, Unsubscribable } from "rxjs"
 import { WrappedMessage } from "../utils/message.ordering"
 
 /* EVENT BUS */
@@ -14,42 +14,48 @@ export interface GeneralEvent<T> {
 
 
 /* MANAGEMENT */
-export interface MessageTransmissionManagerInterface {
-    getTransmitter(profile: TransmissionProfile): MessageTransmissionInterface
-    getReceiver(profile: TransmissionProfile): MessageReceiverInterface
+export interface MessageTransmissionManagerInterface<T> {
+    getTransmissionObject(config: TransmissionConfig, role: TransmissionRole): MessageTransmissionInterface<T> | null
+
+    // Previous version
+    getTransmitter(config: TransmissionConfig): MessageTransmissionInterface<T>
+    getReceiver(config: TransmissionConfig): MessageReceiverInterface<T>
 }
 
 export interface AdapterManagerInterface {
-    subscribeForAdapters(receiverId: string, role: TransmissionRole): Observable<AdapterInterface>
+    subscribeForAdapters(): Observable<AdapterInterface<any>>
 }
 
 
 /* TRANSMISSION COMPONENTS */
-export interface MessageTransmissionInterface {
+export interface MessageTransmissionInterface<T> {
+    getTransmissionProfile(type?: `config` | `id` | `role`): TransmissionProfile | TransmissionConfig | string | undefined
+    subscribe(observer: Observer<T> | Observable<T>): Unsubscribable | null
 }
 
-export interface MessageReceiverInterface extends MessageTransmissionInterface {
+export interface MessageReceiverInterface<T> extends MessageTransmissionInterface<T> {
     getReceivables(): Observable<any>
 }
 
-export interface MessageTransmitterInterface extends MessageTransmissionInterface {
+export interface MessageTransmitterInterface<T> extends MessageTransmissionInterface<T> {
     emit(message: any): void
 }
 
-export interface MessageRequestResponseInterface extends MessageTransmissionInterface {
-    send(message: any): Observable<any> 
+export interface MessageRequestResponseInterface<T> extends MessageTransmissionInterface<T> {
+    send(message: any): Observable<any>
 }
 
 /* ADAPTER COMPONENTS */
-export interface AdapterInterface {
+export interface AdapterInterface<T> {
+    subscribe(observer: Observer<T>): Unsubscribable
     getAdapterProfile(type?: `id` | `clientId` | `role` | `transportId` | `transportType` | `connectionState`): AdapterProfile | string | Observable<ConnectionState> | undefined
 }
 
-export interface TransmitterAdapterInterface extends AdapterInterface {
+export interface TransmitterAdapterInterface<T> extends AdapterInterface<T> {
     emit(selfId: string, message: WrappedMessage): void
 }
 
-export interface ReceiverAdapterInterface extends AdapterInterface {
+export interface ReceiverAdapterInterface<T> extends AdapterInterface<T> {
     subscribeForIncoming(): Observable<any>
 }
 
@@ -75,16 +81,9 @@ export interface FisMessage {
 
 export type TransportType = 'Websocket' | 'Http' | 'TCP' | undefined
 
-export enum AdapterTransmissionRole {
-    Transmitter,
-    Receiver,
-    RequestResponse
-}
-
-
 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' | 'Adapter Manager Started' | 'Available Transport' | 'Pushing Adapter'
+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 {
@@ -98,10 +97,6 @@ export interface TransportServiceProfile {
     transport: TransportType
 }
 
-export interface Info {
-    transport: TransportType
-}
-
 export interface ClientObject {
     clientId: string,
     dateCreated: Date,
@@ -113,7 +108,7 @@ export interface TransportSet {
 }
 
 export interface AdapterProfile {
-    id: string, 
+    id: string,
     clientId: string,
     role: TransmissionRole,
     transportType: TransportType,
@@ -122,6 +117,12 @@ export interface AdapterProfile {
 }
 
 export interface TransmissionProfile {
+    id: string
+    role: TransmissionRole
+    config: TransmissionConfig
+}
+
+export interface TransmissionConfig {
     source: string
     target: string
 }

src/test/receiver.ts

@@ -10,11 +10,11 @@ can assume dual roles. So it doesn't really matter.
 
 Please note: For further testing, can just copy and paste from the transmitter.ts, with some tweaks on getting
 profile information as well as setting the isClient boolean.*/
-import { filter, interval, map, Observable, Observer, Subject, Subscription, take } from "rxjs";
+import { filter, interval, map, Observable, Observer, Subject, Subscription, take, Unsubscribable } from "rxjs";
 import { v4 as uuidv4 } from 'uuid'
 import { MessageTransmissionManager } from "../transmission/msg.transmission.manager";
 import ConsoleLogger from "../utils/log.utils";
-import { FisMessage, GeneralEvent, TransportType, TransportServiceInterface, TransportSet, TransmissionProfile, TransportMessage } from "../interface/interface";
+import { FisMessage, GeneralEvent, TransportType, TransportServiceInterface, TransportSet, TransmissionProfile, TransportMessage, TransmissionConfig } from "../interface/interface";
 import config from '../config/config.json';
 import clientProfile from '../../clients/clientprofile.json'
 import { MessageTransmissionReceiver } from "../transmission/msg.transmission.receiver";
@@ -31,7 +31,7 @@ class Supervisor {
     private event!: Subject<GeneralEvent<any>>
     private transportSet: TransportSet[] = []
     private transportServiceArray: TransportServiceInterface[] = []
-    private config: TransmissionProfile = {
+    private config: TransmissionConfig = {
         target: clientProfile.clientId,
         source: clientProfile.id
     }
@@ -47,24 +47,34 @@ class Supervisor {
         this.messageProducer = new MessageProducer(this.clientIncomingMessage)
         this.transmissionManager = new MessageTransmissionManager(this.event, this.isClient)
         this.startMessageTransmission()
+        this.clientIncomingMessage.subscribe(message => this.console.log({ message: `Received ${message.header.messageID ?? `Undefined`}` }))
 
     }
 
     // Testing here. Just comment out or uncomment out the parts you want to test
     private startMessageTransmission(): void {
-        let transmitter: MessageTransmissionTransmitter = this.transmissionManager.getTransmitter(this.config) as MessageTransmissionTransmitter
-        let receiver: MessageTransmissionReceiver = this.transmissionManager.getReceiver(this.config) as MessageTransmissionReceiver
-        let requestResponse: MessageTransmissionRequestResponse = this.transmissionManager.getRequestResponse(this.config, transmitter, receiver)
+        /* Actor Version <All operations are synchronized for now>  [Can set to async at a later time]*/
+        let receiver: MessageTransmissionReceiver<any> = this.transmissionManager.getTransmissionObject(this.config, `Receiver`) as MessageTransmissionReceiver<any>
+        let receiverSubscription = receiver.subscribe(this.clientIncomingMessage as Observer<any>) as Unsubscribable
+        setTimeout(() => {
+            receiverSubscription!.unsubscribe()
+            this.console.log({ message: `Stop receiving messag after 10 seconds` })
+        }, 20000)
+
+        /* Primitive Version */
+        // let transmitter: MessageTransmissionTransmitter<any> = this.transmissionManager.getTransmitter(this.config) as MessageTransmissionTransmitter<any>
+        // let receiver: MessageTransmissionReceiver<any> = this.transmissionManager.getReceiver(this.config) as MessageTransmissionReceiver<any>
+        // let requestResponse: MessageTransmissionRequestResponse<any> = this.transmissionManager.getRequestResponse(this.config, transmitter, receiver)
 
         // emit Message only
-        this.emitMessage(transmitter, this.messageProducer.getNotificationMessage())
+        // this.emitMessage(transmitter, this.messageProducer.getNotificationMessage())
         // this.emitMessage(transmitter, this.messageProducer.getOutgoingMessages())
 
         // receive Message only
-        this.streamMessage(receiver).subscribe((message: FisMessage) => {
-            this.console.log({ message: `Received message ${message.header.messageID}` })
-            this.clientIncomingMessage.next(message)
-        })
+        // this.streamMessage(receiver).subscribe((message: FisMessage) => {
+        //     this.console.log({ message: `Received message ${message.header.messageID}` })
+        //     this.clientIncomingMessage.next(message)
+        // })
 
         // request-response emulation
         // this.sendMessage(requestResponse, { header: { messageID: `123`, messageName: `RequestMessage` }, data: `Test Data` } as FisMessage).subscribe({
@@ -74,11 +84,11 @@ class Supervisor {
         // })
     }
 
-    private emitMessage(transmitter: MessageTransmissionTransmitter, source: Subject<any> | Observable<any>): void {
+    private emitMessage(transmitter: MessageTransmissionTransmitter<any>, source: Subject<any> | Observable<any>): void {
         source.subscribe(message => transmitter.emit(message))
     }
 
-    private sendMessage(requestResponse: MessageTransmissionRequestResponse, message: any): Observable<any> {
+    private sendMessage(requestResponse: MessageTransmissionRequestResponse<any>, message: any): Observable<any> {
         return new Observable((response: Observer<any>) => {
             requestResponse.send(message).subscribe({
                 next: respMessage => {
@@ -97,7 +107,7 @@ class Supervisor {
     }
 
     // Just for clarification, this is incoming not outgoing stream. Apologies in advanced for the lack of a better term.
-    private streamMessage(receiverInstance: MessageTransmissionReceiver): Observable<any> {
+    private streamMessage(receiverInstance: MessageTransmissionReceiver<any>): Observable<any> {
         return new Observable((response: Observer<any>) => {
             const subscription: Subscription = receiverInstance.getReceivables().subscribe({
                 next: message => {

src/test/transmitter.ts

@@ -11,7 +11,7 @@ import { filter, interval, map, Observable, Observer, Subject, Subscription, tak
 import { v4 as uuidv4 } from 'uuid'
 import { MessageTransmissionManager } from "../transmission/msg.transmission.manager";
 import ConsoleLogger from "../utils/log.utils";
-import { FisMessage, GeneralEvent, TransportType, TransportServiceInterface, TransportSet, TransmissionProfile, TransportMessage } from "../interface/interface";
+import { FisMessage, GeneralEvent, TransportType, TransportServiceInterface, TransportSet, TransmissionProfile, TransportMessage, TransmissionConfig } from "../interface/interface";
 import serverProfile from '../../clients/serverprofile.json'
 import { MessageTransmissionReceiver } from "../transmission/msg.transmission.receiver";
 import { MessageTransmissionTransmitter } from "../transmission/msg.transmission.transmitter";
@@ -26,7 +26,7 @@ class Supervisor {
     private event!: Subject<GeneralEvent<any>>
     private transportSet: TransportSet[] = []
     private transportServiceArray: TransportServiceInterface[] = []
-    private config: TransmissionProfile = {
+    private config: TransmissionConfig = {
         target: serverProfile.clientId,
         source: serverProfile.id
     }
@@ -47,19 +47,28 @@ class Supervisor {
 
     // Testing here. Just comment out or uncomment out the parts you want to test
     private startMessageTransmission(): void {
-        let transmitter: MessageTransmissionTransmitter = this.transmissionManager.getTransmitter(this.config) as MessageTransmissionTransmitter
-        let receiver: MessageTransmissionReceiver = this.transmissionManager.getReceiver(this.config) as MessageTransmissionReceiver
-        let requestResponse: MessageTransmissionRequestResponse = this.transmissionManager.getRequestResponse(this.config, transmitter, receiver)
+        /* Actor Version <All operations are synchronized for now>  [Can set to async at a later time]*/
+        let transmitter: MessageTransmissionTransmitter<any> = this.transmissionManager.getTransmissionObject(this.config, `Transmitter`) as MessageTransmissionTransmitter<any>
+        let transmitterSubscription = transmitter.subscribe(this.messageProducer.getNotificationMessage())
+        setTimeout(() => {
+            transmitterSubscription!.unsubscribe()
+            this.console.log({ message: `Stop transmitting messag after 10 seconds` })
+        }, 20000)
+
+        /* Primitive Version */
+        // let transmitter: MessageTransmissionTransmitter<any> = this.transmissionManager.getTransmitter(this.config) as MessageTransmissionTransmitter<any>
+        // let receiver: MessageTransmissionReceiver<any> = this.transmissionManager.getReceiver(this.config) as MessageTransmissionReceiver<any>
+        // let requestResponse: MessageTransmissionRequestResponse<any> = this.transmissionManager.getRequestResponse(this.config, transmitter, receiver)
 
         // emit Message only
-        this.emitMessage(transmitter, this.messageProducer.getNotificationMessage())
+        // this.emitMessage(transmitter, this.messageProducer.getNotificationMessage())
         // this.emitMessage(transmitter, this.messageProducer.getOutgoingMessages())
 
         // receive Message only
-        this.streamMessage(receiver).subscribe((message: FisMessage) => {
-            this.console.log({ message: `Received message ${message.header.messageID}` })
-            this.clientIncomingMessage.next(message)
-        })
+        // this.streamMessage(receiver).subscribe((message: FisMessage) => {
+        //     this.console.log({ message: `Received message ${message.header.messageID}` })
+        //     this.clientIncomingMessage.next(message)
+        // })
 
         // request-response emulation
         // this.sendMessage(requestResponse, { header: { messageID: `123`, messageName: `RequestMessage` }, data: `Test Data` } as FisMessage).subscribe({
@@ -69,11 +78,11 @@ class Supervisor {
         // })
     }
 
-    private emitMessage(transmitter: MessageTransmissionTransmitter, source: Subject<any> | Observable<any>): void {
+    private emitMessage(transmitter: MessageTransmissionTransmitter<any>, source: Subject<any> | Observable<any>): void {
         source.subscribe(message => transmitter.emit(message))
     }
 
-    private sendMessage(requestResponse: MessageTransmissionRequestResponse, message: any): Observable<any> {
+    private sendMessage(requestResponse: MessageTransmissionRequestResponse<any>, message: any): Observable<any> {
         return new Observable((response: Observer<any>) => {
             requestResponse.send(message).subscribe({
                 next: respMessage => {
@@ -92,7 +101,7 @@ class Supervisor {
     }
 
     // Just for clarification, this is incoming not outgoing stream. Apologies in advanced for the lack of a better term.
-    private streamMessage(receiverInstance: MessageTransmissionReceiver): Observable<any> {
+    private streamMessage(receiverInstance: MessageTransmissionReceiver<any>): Observable<any> {
         return new Observable((response: Observer<any>) => {
             const subscription: Subscription = receiverInstance.getReceivables().subscribe({
                 next: message => {

src/transmission/msg.transmission.manager.ts

@@ -5,38 +5,69 @@ import { MessageTransmissionReceiver } from "./msg.transmission.receiver";
 import { AdapterManager } from "../adapters/adapter.manager";
 import { v4 as uuidv4 } from 'uuid'
 import { MessageTransmissionRequestResponse } from "./msg.transmission.request-response";
-import { Subject } from "rxjs";
+import { filter, map, Subject } from "rxjs";
 import ConsoleLogger from "../utils/log.utils";
-import { GeneralEvent, MessageReceiverInterface, MessageTransmitterInterface, TransmissionProfile } from "../interface/interface";
+import { AdapterInterface, AdapterManagerInterface, GeneralEvent, MessageReceiverInterface, MessageTransmissionInterface, MessageTransmitterInterface, TransmissionConfig, TransmissionProfile, TransmissionRole } from "../interface/interface";
 import { MessageTransmissionManagerBase } from "../base/msg.transmission.manager.base";
 
-export class MessageTransmissionManager extends MessageTransmissionManagerBase {
+export class MessageTransmissionManager extends MessageTransmissionManagerBase<any> {
 
-    constructor(event: Subject<GeneralEvent<any>>, browserEnv?: boolean) {
+    constructor(event: Subject<GeneralEvent<any>>, browserEnv?: boolean, adapterManager?: AdapterManager) {
         super()
         this.console = new ConsoleLogger(`MessageTransmissionManager`, ['managers'])
         if (browserEnv) this.browserEnv = browserEnv
         this.console.log({ message: `Contructing self... ${this.browserEnv ? `is receiving end` : `is not browser env`}` })
-        this.event = event
-        // Instantiate adapter manager
-        this.adapterManager = new AdapterManager(event, this.browserEnv)
+        this.globalEvent = event
+        this.transmissionEvent = new Subject<any>()
+        this.adapterManager = adapterManager ? adapterManager : new AdapterManager(event, this.browserEnv)
+        this.setupTransmissionEvent(this.adapterManager)
     }
 
-    // for now these fuctions ain't pure, cuz messageTransmissionManager needs to keep a record of these tranmission instances as well for safe keeping.
-    public getTransmitter(profile: TransmissionProfile): MessageTransmitterInterface {
-        let transmitterInstance = new MessageTransmissionTransmitter(profile, this.adapterManager)
-        this.transmissionRef.push(transmitterInstance)
+    public getTransmissionObject(config: TransmissionConfig, role: TransmissionRole): MessageTransmissionInterface<any> | null {
+        if (role === `Transmitter`) return this.getTransmitter(config)
+        if (role === `Receiver`) return this.getReceiver(config)
+        else {
+            let transmitter = this.transmissionObjects.find(obj => (obj.getTransmissionProfile() as TransmissionProfile).config.target === config.target && (obj.getTransmissionProfile() as TransmissionProfile).role === 'Transmitter')
+            let receiver = this.transmissionObjects.find(obj => (obj.getTransmissionProfile() as TransmissionProfile).config.target === config.target && (obj.getTransmissionProfile() as TransmissionProfile).role === 'Receiver')
+            if (transmitter && receiver) {
+                return this.getRequestResponse(config, transmitter as MessageTransmitterInterface<any>, receiver as MessageReceiverInterface<any>)
+            } else {
+                return null
+            }
+        }
+    }
+
+
+    public getTransmitter(config: TransmissionConfig): MessageTransmitterInterface<any> {
+        let transmitterInstance = new MessageTransmissionTransmitter(config, this.transmissionEvent)
+        this.transmissionObjects.push(transmitterInstance)
         return transmitterInstance
     }
 
-    public getReceiver(profile: TransmissionProfile): MessageReceiverInterface {
-        let receiverInstance = new MessageTransmissionReceiver(profile, this.adapterManager)
-        this.transmissionRef.push(receiverInstance)
+    public getReceiver(config: TransmissionConfig): MessageReceiverInterface<any> {
+        let receiverInstance = new MessageTransmissionReceiver(config, this.transmissionEvent)
+        this.transmissionObjects.push(receiverInstance)
         return receiverInstance
     }
 
-    public getRequestResponse(profile: TransmissionProfile, transmitterInstance: MessageTransmitterInterface, receiverInstance: MessageReceiverInterface): MessageTransmissionRequestResponse {
-        return new MessageTransmissionRequestResponse(profile, transmitterInstance, receiverInstance)
+    public getRequestResponse(config: TransmissionConfig, transmitterInstance: MessageTransmitterInterface<any>, receiverInstance: MessageReceiverInterface<any>): MessageTransmissionRequestResponse<any> {
+        return new MessageTransmissionRequestResponse(config, transmitterInstance, receiverInstance)
+    }
+
+    // Subscribe for new adapters from Adapter Manager
+    private setupTransmissionEvent(adapterManager: AdapterManagerInterface): void {
+        adapterManager.subscribeForAdapters().pipe(
+            map((adapter: AdapterInterface<any>) => {
+                return {
+                    id: uuidv4(),
+                    type: `Transmission Event`,
+                    event: `New Adapter`,
+                    date: new Date(),
+                    data: adapter,
+                    transport: adapter.getAdapterProfile(`transportType`)
+                } as GeneralEvent<AdapterInterface<any>>
+            })
+        ).subscribe(this.transmissionEvent)
     }
 
 }

src/transmission/msg.transmission.receiver.ts

@@ -6,29 +6,39 @@ i) Logic to dynamically switch adapters, either based on their connection status
 ii) Enabling the use of mutli adapters usage to increase bandwith for data transmission. (More Advanced)
 
  */
-import { BehaviorSubject, filter, Observable, Observer, Subject, Subscription } from 'rxjs';
+import { BehaviorSubject, filter, map, Observable, Observer, Subject, Subscription, Unsubscribable } from 'rxjs';
 import { checkMessage, WrappedMessage } from '../utils/message.ordering';
 import ConsoleLogger from '../utils/log.utils';
 import { MessageTransmissionBase } from '../base/msg.transmission.base';
-import { AdapterInterface, AdapterManagerInterface, ConnectionState, FisMessage, GeneralEvent, MessageReceiverInterface, ReceiverAdapterInterface, TransmissionProfile, TransportMessage } from '../interface/interface';
+import { AdapterInterface, AdapterManagerInterface, ConnectionState, FisMessage, GeneralEvent, MessageReceiverInterface, ReceiverAdapterInterface, TransmissionConfig, TransmissionProfile, TransportMessage } from '../interface/interface';
+import { checkRxType } from '../utils/general.utils';
 
-export class MessageTransmissionReceiver extends MessageTransmissionBase implements MessageReceiverInterface {
+export class MessageTransmissionReceiver<T> extends MessageTransmissionBase<T> {
     private connectionStateEvent: BehaviorSubject<ConnectionState> = new BehaviorSubject<ConnectionState>(`OFFLINE`)
     private console: ConsoleLogger = new ConsoleLogger(`MessageTransmissionReceiver`, ['transmission'])
     private onHoldMessage: Subject<WrappedMessage> = new Subject()
-    private currentAdapter!: ReceiverAdapterInterface
+    private currentAdapter!: ReceiverAdapterInterface<any>
     private incomingMessage: Subject<GeneralEvent<TransportMessage>> = new Subject()
 
-    constructor(profile: TransmissionProfile, adapterManager: AdapterManagerInterface) {
-        super()
-        this.profile = profile
-        this.console.log({ message: `Constructing Receiver Transmission for Receiving target: ${this.profile.target}` })
-        this.initializeReceiverComponents(adapterManager)
+    constructor(config: TransmissionConfig, transmissionEvent: Observable<GeneralEvent<any>>) {
+        super(config, transmissionEvent)
+        this.setTransmissionProfile(`Receiver`, config)
+        this.console.log({ message: `Constructing Receiver Transmission for Receiving target: ${this.transmissionProfile.config.target}` })
+        this.initializeReceiverComponents(transmissionEvent)
+    }
+
+    public subscribe(param: Observer<any> | Observable<any>): Unsubscribable | null {
+        if (checkRxType(param) === `Observer`) {
+            this.console.log({ message: `Is Observer` });
+            return this.getReceivables().subscribe(param as Observer<any>)
+        } else {
+            return null
+        }
     }
 
     public getReceivables(): Observable<FisMessage> {
         return new Observable((receivable: Observer<FisMessage>) => {
-            this.console.log({ message: `Tranmission Subscription: Streaming incoming messages from ${this.profile.target}` })
+            this.console.log({ message: `Tranmission Subscription: Streaming incoming messages from ${this.transmissionProfile.config.target}` })
             const subscription: Subscription = this.incomingMessage.pipe(
                 filter((event: GeneralEvent<any>) => event.type == `Adapter Event`),
                 filter((event: GeneralEvent<any>) => event.event == 'New Message'),
@@ -52,14 +62,19 @@ export class MessageTransmissionReceiver extends MessageTransmissionBase impleme
     }
 
     /* Assigned and update adapters record. Currently no logic to swtich adapters based on performance or whatever logic to be integrated in the future */
-    private initializeReceiverComponents(adapterManager: AdapterManagerInterface): void {
-        adapterManager.subscribeForAdapters(this.profile.target, `Receiver`).pipe(
-        ).subscribe((adapter: AdapterInterface) => {
+    private initializeReceiverComponents(transmissionEvent: Observable<GeneralEvent<AdapterInterface<any>>>): void {
+        transmissionEvent.pipe(
+            filter(event => event.type === `Transmission Event`),
+            filter(event => event.event === `New Adapter`),
+            filter(event => (event.data as AdapterInterface<any>).getAdapterProfile(`clientId`) === this.transmissionProfile.config.target),
+            filter(event => (event.data as AdapterInterface<any>).getAdapterProfile(`role`) === `Receiver`),
+            map(event => { return event.data as AdapterInterface<any> })
+        ).subscribe((adapter: AdapterInterface<any>) => {
             this.adapters.push(adapter)
             this.console.log({ message: `Adding new ${adapter.getAdapterProfile(`transportType`)} receiving adapter. Current adapter length: ${this.adapters.length}` })
             if (!this.currentAdapter) {
                 this.console.log({ message: `Setting this ${adapter.getAdapterProfile(`id`)} as current adapter.` })
-                this.currentAdapter = adapter as ReceiverAdapterInterface
+                this.currentAdapter = adapter as ReceiverAdapterInterface<any>
                 this.currentAdapter.subscribeForIncoming().subscribe({
                     next: (message: GeneralEvent<TransportMessage>) => {
                         this.console.log({ message: `Received ${(((message.data as TransportMessage).payload as WrappedMessage).payload as FisMessage).header.messageID} from ${((message.data as TransportMessage).source)}`, details: message })

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

@@ -2,23 +2,43 @@
 mechanism is still using message. Here as you will see, it is basically taking the already instantiated transmitter and receiver components
 and basically just filtering the responses based on whatever identifier it needs for the orignal request. */
 import { MessageTransmissionBase } from "../base/msg.transmission.base";
-import { filter, Observable, Observer, Subscription } from "rxjs";
-import { FisMessage, MessageReceiverInterface, MessageRequestResponseInterface, MessageTransmitterInterface, TransmissionProfile } from "../interface/interface";
+import { filter, Observable, Observer, Subscription, Unsubscribable } from "rxjs";
+import { FisMessage, MessageReceiverInterface, MessageRequestResponseInterface, MessageTransmitterInterface, TransmissionConfig, TransmissionProfile } from "../interface/interface";
 import ConsoleLogger from "../utils/log.utils";
+import { checkRxType } from "../utils/general.utils";
 
-export class MessageTransmissionRequestResponse extends MessageTransmissionBase implements MessageRequestResponseInterface {
+export class MessageTransmissionRequestResponse<T> implements MessageRequestResponseInterface<T> {
     private console: ConsoleLogger = new ConsoleLogger(`MessageTransmissionRequestResponse`, ['transmission'])
-    protected transmitterInstance!: MessageTransmitterInterface;
-    protected receiverInstance!: MessageReceiverInterface;
+    private transmissionProfile!: TransmissionProfile
+    protected transmitterInstance!: MessageTransmitterInterface<any>;
+    protected receiverInstance!: MessageReceiverInterface<any>;
 
-    constructor(profile: TransmissionProfile, transmitterInstance: MessageTransmitterInterface, receiverInstance: MessageReceiverInterface) {
-        super()
-        this.profile = profile
-        this.console.log({ message: `Constructing Request Response Transmission for Receiving target: ${this.profile.target}` })
+    constructor(config: TransmissionConfig, transmitterInstance: MessageTransmitterInterface<any>, receiverInstance: MessageReceiverInterface<any>) {
+        this.console.log({ message: `Constructing Request Response Transmission for Receiving target: ${config.target}` })
         this.transmitterInstance = transmitterInstance
         this.receiverInstance = receiverInstance
     }
 
+    getTransmissionProfile(type?: `config` | `id` | `role`): TransmissionProfile | TransmissionConfig | string | undefined {
+        return type ? this.transmissionProfile[type] : this.transmissionProfile;
+    }
+
+    // this is still broken at this point in time. DO NOT USE!!!!
+    subscribe(observer: Observer<T> | Observable<T>): Unsubscribable | null {
+        if (checkRxType(observer) === `Observable`) {
+            // Create a new Subscription to manage unsubscription
+            const subscription = (observer as Observable<any>).subscribe(message => {
+                this.send(message).subscribe(response => {
+                    this.console.log({ message: `Response received: ${response}` });
+                });
+            });
+
+            return subscription; // Return the Subscription (Unsubscribable)
+        } else {
+            return null;
+        }
+    }
+
     send(message: FisMessage): Observable<FisMessage> {
         return new Observable((response: Observer<FisMessage>) => {
             // logic here

src/transmission/msg.transmission.transmitter.ts

@@ -8,29 +8,44 @@ ii) Enabling the use of mutli adapters usage to increase bandwith for data trans
 */
 import { MessageTransmissionBase } from "../base/msg.transmission.base";
 import { v4 as uuidv4 } from 'uuid'
-import { BehaviorSubject, distinctUntilChanged, filter, map, Observable, Subject, Subscription } from "rxjs";
+import { BehaviorSubject, distinctUntilChanged, filter, map, Observable, Observer, Subject, Subscription, Unsubscribable } from "rxjs";
 import { RetransmissionService } from "../utils/retransmission.service";
 import { WrappedMessage } from "../utils/message.ordering";
 import ConsoleLogger from "../utils/log.utils";
-import { AdapterInterface, AdapterManagerInterface, ConnectionState, FisMessage, GeneralEvent, MessageTransmitterInterface, TransmissionProfile, TransmitterAdapterInterface, TransportMessage } from "../interface/interface";
-import { error } from "console";
-import { TransmitterAdapter } from "../adapters/adapter.transmitter";
+import { AdapterInterface, AdapterManagerInterface, ConnectionState, FisMessage, GeneralEvent, MessageTransmitterInterface, TransmissionConfig, TransmissionProfile, TransmitterAdapterInterface, TransportMessage } from "../interface/interface";
+import { checkRxType } from "../utils/general.utils";
 
 
-export class MessageTransmissionTransmitter extends MessageTransmissionBase implements MessageTransmitterInterface {
+export class MessageTransmissionTransmitter<T> extends MessageTransmissionBase<T> {
+    private internalObservable: Observable<GeneralEvent<T>> = new Observable()
     private connectionStateEvent: BehaviorSubject<ConnectionState> = new BehaviorSubject<ConnectionState>('OFFLINE')
     private console: ConsoleLogger = new ConsoleLogger(`MessageTransmissionTransmitter`, ['transmission'])
     private messageToBeBuffered!: Subject<FisMessage | WrappedMessage>
     private buffer!: RetransmissionService;
-    private currentAdapter!: TransmitterAdapterInterface
+    private currentAdapter!: TransmitterAdapterInterface<any>
 
-    constructor(profile: TransmissionProfile, adapterManager: AdapterManagerInterface) {
-        super()
-        this.profile = profile
-        this.console.log({ message: `Constructing Transmitter Transmission for Receiving target: ${this.profile.target}` })
+    constructor(config: TransmissionConfig, transmissionEvent: Observable<GeneralEvent<any>>) {
+        super(config, transmissionEvent)
+        this.setTransmissionProfile(`Transmitter`, config)
+        this.console.log({ message: `Constructing Transmitter Transmission for Receiving target: ${this.transmissionProfile.config.target}` })
         this.messageToBeBuffered = new Subject()
         this.buffer = new RetransmissionService()
-        this.initializeTransmitterComponents(adapterManager)
+        this.initializeTransmitterComponents(transmissionEvent)
+    }
+
+    public subscribe(observer: Observer<any> | Observable<any>): Unsubscribable | null {
+        if (checkRxType(observer) === `Observable`) {
+            this.console.log({ message: `Is Observable` });
+            // Create a new Subscription to manage unsubscription
+            const subscription = (observer as Observable<any>).subscribe(message => {
+                this.emit(message);
+                this.console.log({ message: `Message ${message.header?.messageID ?? `Undefined`} being processed... ` });
+            });
+
+            return subscription; // Return the Subscription (Unsubscribable)
+        } else {
+            return null;
+        }
     }
 
     public emit(message: FisMessage): void {
@@ -40,15 +55,21 @@ export class MessageTransmissionTransmitter extends MessageTransmissionBase impl
 
     /* After setting up, will listen specifically to the connection state of this particular remote client. So that, the buffer signal can be
     established to allow the buffer to do their thing. */
-    private initializeTransmitterComponents(adapterManager: AdapterManagerInterface): void {
+    private initializeTransmitterComponents(transmissionEvent: Observable<GeneralEvent<AdapterInterface<any>>>): void {
         this.console.log({ message: `Setting up Retransmission Service...` })
         // Listen and update adapters
-        adapterManager.subscribeForAdapters(this.profile.target, `Transmitter`).subscribe((adapter: AdapterInterface) => {
+        transmissionEvent.pipe(
+            filter(event => event.type === `Transmission Event`),
+            filter(event => event.event === `New Adapter`),
+            filter(event => (event.data as AdapterInterface<any>).getAdapterProfile(`clientId`) === this.transmissionProfile.config.target),
+            filter(event => (event.data as AdapterInterface<any>).getAdapterProfile(`role`) === `Transmitter`),
+            map(event => { return event.data as AdapterInterface<any> })
+        ).subscribe((adapter: AdapterInterface<any>) => {
             this.adapters.push(adapter)
             this.console.log({ message: `Adding new ${adapter.getAdapterProfile(`transportType`)} transmitting adapter. Current adapter length: ${this.adapters.length}` })
             if (!this.currentAdapter) {
                 this.console.log({ message: `Setting this ${adapter.getAdapterProfile(`id`)} as current adapter.` })
-                this.currentAdapter = adapter as TransmitterAdapterInterface
+                this.currentAdapter = adapter as TransmitterAdapterInterface<T>
                 let connectionState: Observable<ConnectionState> = this.currentAdapter.getAdapterProfile('connectionState') as Observable<ConnectionState>
                 connectionState.subscribe(this.connectionStateEvent)
             }
@@ -60,7 +81,7 @@ export class MessageTransmissionTransmitter extends MessageTransmissionBase impl
             // need to work with wrapped messages
             this.console.log({ message: `Transmitting ${bufferedMessage.thisMessageID}` });
             if (this.currentAdapter) {
-                this.currentAdapter.emit(this.profile.source, bufferedMessage)
+                this.currentAdapter.emit(this.transmissionProfile.config.source, bufferedMessage)
             } else {
                 // just flush back the message inside the buffer, if the adapter is not ready or assigned.
                 this.messageToBeBuffered.next(bufferedMessage)

src/utils/general.utils.ts

@@ -2,6 +2,7 @@ import * as fs from 'fs'
 import path from 'path';
 import ConsoleLogger from './log.utils';
 import { ClientObject, TransportServiceInterface } from '../interface/interface';
+import { isObservable, Observable, Observer, Subject } from 'rxjs';
 const console: ConsoleLogger = new ConsoleLogger(`GeneralUtils`, ['util'])
 // Check if filename exists. Return profile information if there's any
 export async function checkOwnClientInfo(filename?: string): Promise<{ id: string, clientId: string }> {
@@ -116,3 +117,27 @@ export function addClientToDB(entry: any, filename: string): void {
     }
 }
 
+function isRxObservable(value: any): value is Observable<any> {
+    return isObservable(value);
+}
+function isRxObserver(value: any): value is Observer<any> {
+    return (
+        value &&
+        typeof value === 'object' &&
+        typeof value.next === 'function' &&
+        typeof value.error === 'function' &&
+        typeof value.complete === 'function'
+    );
+}
+
+// Check specifically if the value is a Subject
+function isRxSubject(value: any): value is Subject<any> {
+    return isRxObservable(value) && isRxObserver(value);
+}
+
+export function checkRxType(value: any): 'Subject' | 'Observable' | 'Observer' | 'Neither' {
+    if (isRxSubject(value)) return 'Observer'
+    if (isRxObservable(value)) return 'Observable'
+    if (isRxObserver(value)) return 'Observer'
+    return 'Neither';
+}