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 = new Subject() private currentAdapter!: ReceiverAdapterInterface // private toBePassedOver: Subject = new Subject() constructor(clientId: string, eventObj: EventObject) { super() this.clientId = clientId this.eventObj = eventObj this.handleAdapterEvent(this.eventObj.adapterEvent.asObservable()) } getIncoming(): Observable { this.console.log({ message: `Transmission getting message bus for ${this.clientId}` }) return new Observable((observable: Observer) => { // 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): void { } }