Fis-MessageTransmission/src/transmission/msg.transmission.receiver.ts

import { filter, Observable, Observer, Subject, Subscription } from 'rxjs';
import { v4 as uuidv4 } from 'uuid'
import { ReceiverAdapter } from '../connector/adapter.receiver';
import { checkMessage, WrappedMessage } from '../utils/message.ordering';
import ConsoleLogger from '../utils/log.utils';
import { MessageTransmissionBase } from '../base/msg.transmission.base';
import { MessageReceiverInterface, ReceiverAdapterInterface, TransportMessage } from '../interface/interface';

export class MessageTransmissionReceiver extends MessageTransmissionBase implements MessageReceiverInterface {
    private console: ConsoleLogger = new ConsoleLogger(`MessageTransmissionReceiver`, ['transmission'])
    private onHoldMessage: Subject<WrappedMessage> = new Subject()
    private currentAdapter!: ReceiverAdapterInterface
    // private toBePassedOver: Subject<WrappedMessage> = new Subject()

    constructor(clientId: string, eventObj: EventObject) {
        super()
        this.clientId = clientId
        this.eventObj = eventObj

        this.handleAdapterEvent(this.eventObj.adapterEvent.asObservable())
    }

    getIncoming(): Observable<TransportEvent> {
        this.console.log({ message: `Transmission getting message bus for ${this.clientId}` })
        return new Observable((observable: Observer<TransportEvent>) => {
            // logic here
            // Need to merge all the adapters into one when the time comes 
            // SAMPLE: This adapterArray.forEach(adapter => { ... })
            const subscription: Subscription = this.currentAdapter.subscribe().pipe(
                filter((event: TransportEvent) => event.event == 'New Message'),
            ).subscribe((event: TransportEvent) => {
                // console.log(event) // data is transportMessage instead of eventmessage
                this.onHoldMessage.next(((event.data as TransportMessage).payload as WrappedMessage))
                checkMessage(((event.data as TransportMessage).payload as WrappedMessage), this.onHoldMessage).then(() => {
                    // only release the message before it exists
                    this.console.log({ message: `This one passes. Does have previousID. Case for message ordering` })
                    // console.log(((event.data as TransportMessage).payload as WrappedMessage))
                    observable.next(event);
                }).catch((error) => {
                    this.console.log({ message: `Observer Error`, details: error })
                })
            });

            // Clean up on unsubscription
            return () => {
                subscription.unsubscribe();
            };

        })
    }

    private handleAdapterEvent(adapterEvent: Observable<AdapterEvent>): void {

    }

}