retransmission.service.ts 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132
  1. import { BehaviorSubject, buffer, concatMap, distinctUntilChanged, from, Observable, Subject, takeWhile } from "rxjs";
  2. import { BaseMessage } from "../dependencies/logging/dependencies/msgutil/dependencies/dependencies";
  3. import { RetransmissionInterface } from "../interfaces/retransmission.interface";
  4. import { WrappedMessage } from "../interfaces/general.interface";
  5. import { sortMessageBasedOnDate } from "./utility/message-ordering";
  6. import { v4 as uuidV4 } from 'uuid';
  7. export class RetransmissionService implements RetransmissionInterface {
  8. private currentMessageId: string | null
  9. private sortMessage: boolean = false
  10. private bufferReleaseSignal: Subject<void> = new Subject()
  11. private receiverConnectionState: BehaviorSubject<'OFFLINE' | 'ONLINE'> = new BehaviorSubject('OFFLINE')
  12. private transmissionState: BehaviorSubject<'TRANSMITTING' | 'IDLE' | 'ARRAY EMPTY' | 'STORING DATA' | 'GETTING STORED DATA'> = new BehaviorSubject('ARRAY EMPTY')
  13. private arrayToBeTransmitted: Subject<WrappedMessage[]> = new Subject()
  14. private toBeWrapped: Subject<any> = new Subject()
  15. private wrappedMessageToBeBuffered: Subject<WrappedMessage> = new Subject()
  16. private messageToBeTransmitted: Subject<WrappedMessage> = new Subject()
  17. // Interface
  18. public retransmission(payloadToBeTransmitted: Observable<any>, eventListener: Observable<any>, messageOrdering?: boolean) {
  19. if (messageOrdering) {
  20. this.sortMessage = true
  21. console.log(`Message ordering is set to ${this.sortMessage}`)
  22. }
  23. eventListener.subscribe(event => this.receiverConnectionState.next(event))
  24. this.startWrappingOperation()
  25. this.startBufferTransmisionProcess()
  26. this.releaseSignalManager()
  27. payloadToBeTransmitted.subscribe((message) => {
  28. this.toBeWrapped.next(message)
  29. })
  30. }
  31. public returnBufferedMessages(): Observable<WrappedMessage> {
  32. return this.messageToBeTransmitted.asObservable()
  33. }
  34. private startWrappingOperation() {
  35. this.toBeWrapped.subscribe(message => {
  36. this.wrappedMessageToBeBuffered.next(this.wrapMessageWithTimeReceived(message, this.currentMessageId ? this.currentMessageId : null))
  37. })
  38. //simulate connection test
  39. // wrappedMessageToBeBuffered will then be pushed to buffer
  40. this.wrappedMessageToBeBuffered.pipe(buffer(this.bufferReleaseSignal)).subscribe((bufferedMessages: WrappedMessage[]) => {
  41. // console.log(bufferedMessages.length + ' buffered messages')
  42. // console.log(`Released buffered message: ${bufferedMessages.length} total messages. To Be sorted.`)
  43. // arrayToBeTransmitted.next(sortMessage(bufferedMessages))
  44. this.arrayToBeTransmitted.next((this.sortMessage && bufferedMessages.length > 0) ? sortMessageBasedOnDate(bufferedMessages) : bufferedMessages)
  45. });
  46. }
  47. private wrapMessageWithTimeReceived(message: any, previousMessageID: string | null): WrappedMessage {
  48. // check if message has already a time received property if so no need to add anymore
  49. if (!message.timeReceived) {
  50. let WrappedMessage: WrappedMessage = {
  51. timeReceived: new Date(),
  52. payload: message as BaseMessage,
  53. thisMessageID: uuidV4(),
  54. previousMessageID: previousMessageID
  55. }
  56. // console.log(`Current`, WrappedMessage.thisMessageID, 'Previous for this message:', WrappedMessage.previousMessageID)
  57. this.currentMessageId = WrappedMessage.thisMessageID
  58. // console.log(`Updating: `, this.currentMessageId)
  59. return WrappedMessage
  60. } else {
  61. return message as WrappedMessage
  62. }
  63. }
  64. private startBufferTransmisionProcess() {
  65. console.log(`StartBufferTransmissionProcess`)
  66. this.arrayToBeTransmitted.subscribe(array => {
  67. if (array.length > 0) {
  68. this.transmissionState.next('TRANSMITTING')
  69. from(array).subscribe({
  70. next: (message: WrappedMessage) => {
  71. if (this.receiverConnectionState.getValue() == 'OFFLINE') {
  72. // buffer this message. Flush it back to buffer
  73. this.wrappedMessageToBeBuffered.next(message)
  74. }
  75. if (this.receiverConnectionState.getValue() == 'ONLINE') {
  76. this.messageToBeTransmitted.next(message)
  77. }
  78. },
  79. error: err => console.error(err),
  80. complete: () => {
  81. // update transmission state to indicate this batch is completed
  82. // console.log(`Processing buffered array completed. Changing transmission state to ARRAY EMPTY`);
  83. this.transmissionState.next('ARRAY EMPTY');
  84. if (this.receiverConnectionState.getValue() === 'ONLINE' && this.transmissionState.getValue() === 'ARRAY EMPTY') {
  85. setTimeout(() => {
  86. this.bufferReleaseSignal.next()
  87. }, 1000)
  88. }
  89. // Do nothing if the receiver connection is offline
  90. }
  91. });
  92. } else {
  93. // If I don't do setTimeout, then bufferrelasesignal will be overloaded
  94. setTimeout(() => {
  95. this.bufferReleaseSignal.next()
  96. }, 3000)
  97. }
  98. }
  99. )
  100. }
  101. private releaseSignalManager() {
  102. this.receiverConnectionState.pipe(
  103. distinctUntilChanged()
  104. ).subscribe(clientState => {
  105. console.log(`Client is now ${clientState}`)
  106. if (clientState == 'OFFLINE') {
  107. console.log(`Current transmission state: ${this.transmissionState.getValue()}`)
  108. // just keep buffering
  109. }
  110. if (clientState == 'ONLINE') {
  111. console.log(`Current transmission state: ${this.transmissionState.getValue()}`)
  112. // get the stored messages to pump it back into the buffer to be ready to be processed immediately
  113. if (this.transmissionState.getValue() == 'ARRAY EMPTY') {
  114. this.bufferReleaseSignal.next()
  115. }
  116. }
  117. })
  118. }
  119. }