fix buffer

services/buffer.service.ts

@@ -5,10 +5,11 @@ import { ConnectionState, Message, MessageLog } from '../interfaces/general.inte
 import { resolve } from 'path';
 
 export class BufferService {
-    private messageFromApplication: Subject<Message>;
-    private messageFromBuffer: Subject<Message>
+    private messageStream: Subject<Message>
+    // private messageFromApplication: Subject<Message>;
+    // private messageFromBuffer: Subject<Message>
     private connectionState: BehaviorSubject<ConnectionState>
-    private currentSource: Subject<Message>
+    // private currentSource: Subject<Message>
     private messageBuffer: Message[] = [];
     private messageModel: Model<Message> | undefined;
     private readonly dbUrl: string = process.env.MONGO as string;
@@ -19,9 +20,10 @@ export class BufferService {
         connectionStateSubject: BehaviorSubject<ConnectionState>,
         dbName: string
     ) {
-        this.messageFromBuffer = new Subject<Message>();
-        this.messageFromApplication = messageFromApp;
-        this.currentSource = this.messageFromBuffer
+        this.messageStream = messageFromApp;
+        // this.messageFromBuffer = new Subject<Message>();
+        // this.messageFromApplication = messageFromApp;
+        // this.currentSource = this.messageFromBuffer
         this.connectionState = connectionStateSubject;
         this.setupSubscriptions(); // Note: The handle buffer will push the data in local array before pushing to mongo via initial check up model
 
@@ -37,11 +39,11 @@ export class BufferService {
     }
 
     public getMessages(): Observable<Message> {
-        return this.currentSource as Observable<Message>
+        return this.messageStream as Observable<Message>
     }
 
     private setupSubscriptions(): void {
-        this.messageFromApplication.subscribe({
+        this.messageStream.subscribe({
             next: (message: Message) => this.handleIncomingMessage(message),
             error: (err) => console.error('Error in messageToBePublished subject:', err),
             complete: () => console.log('messageToBePublished subscription completed')
@@ -90,23 +92,12 @@ export class BufferService {
     private handleConnectionStateChanges(state: ConnectionState): void {
         console.log(this.connectionState.getValue().status)
         if (state.status === 'BUFFER') {
-            this.currentSource = this.messageFromBuffer
             if (state.payload && typeof state.payload !== 'string') {
                 this.bufferMessage(state.payload); // Buffer the last message immediately
             }
         }
         if (state.status === 'DIRECT_PUBLISH') {
-            if (this.messageBuffer.length > 0) { // temporary only since i am not using mongoDB for buffering atm
-                this.releaseBufferedMessages(this.messageFromBuffer).then(() => {
-                    console.log(`Switching to main publisher from source`)
-                    this.currentSource = this.messageFromApplication
-                }).catch((err) => {
-                    console.error(err)
-                })
-            }
-            if (this.messageBuffer.length < 1) {
-                this.currentSource = this.messageFromApplication
-            }
+            this.releaseBufferedMessages(this.messageStream)
             /* This is for mongo  */
             // this.isBufferNotEmpty().then(isNotEmpty => {
             //     if (isNotEmpty) {
@@ -179,7 +170,7 @@ export class BufferService {
             if (!this.messageModel) {
                 // If MongoDB model is not defined, use the local buffer
                 console.log(`Releasing buffer Message: currently there is ${this.messageBuffer.length} messages to be released`)
-                this.messageBuffer.forEach(message => this.messageFromBuffer.next(message));
+                this.messageBuffer.forEach(message => this.messageStream.next(message));
                 this.messageBuffer.length = 0 // Clear the local buffer after transferring
                 if (this.messageBuffer.length < 1) {
                     resolve(true)

services/grpc.service.method.ts

@@ -138,15 +138,13 @@ export class GrpcServiceMethod {
                         this.callRequestsFromRemote.push(call)
                         let clientInfo = JSON.parse(call.request.message)
                         this.generateAdditionalAttributes(connectionAttribute, clientInfo)
-                        // console.log(`Hello 123`)
-                        // connectionAttribute.outGoing.MessageToBePublished?.subscribe(e => (e.message as MessageLog).appData.msgId)
 
                         console.log(`Initializing stream. Opening Channel... Confirmation from ${call.request.id}`)
 
                         if (connectionAttribute.outGoing.MessageToBePublished) {
                             let subscription: Subscription = connectionAttribute.outGoing.MessageToBePublished.subscribe({
                                 next: (response: Message) => {
-                                    console.log(`Sending from GRPC server: ${(response.message as MessageLog).appData.msgId} `)
+                                    // console.log(`Sending from GRPC server: ${(response.message as MessageLog).appData.msgId} `)
                                     this.messageToBeSendOver = response
                                     let message = {
                                         id: response.id,
@@ -245,6 +243,7 @@ export class GrpcServiceMethod {
             });
 
             call.on('error', (err) => {
+                console.error(err)
                 resolve('')
             });
         })

test/grpc1.ts

@@ -39,7 +39,7 @@ connectionService.generateConnection(connectionRequest)
 
 // let generateFakeMessagesToBePublished = stream().pipe(take(1000))
 
-// let generateFakeMessagesToBePublished = from(parsedMessages).pipe(take(1000))
+// let generateFakeMessagesToBePublished = from(parsedMessages).pipe(take(30000))
 // generateFakeMessagesToBePublished.subscribe({
 //   next: message => {
 //     let payload: Message = {
@@ -316,7 +316,7 @@ function stream(): Subject<any> {
   const intervalId = setInterval(() => {
     result.next(messages[count]);
     count++;
-    if (count >= 1000) {
+    if (count >= 3000) {
       clearInterval(intervalId);
       result.complete();
     }
@@ -342,15 +342,14 @@ function generateFakeStreamResponse(request: any): Subject<any> {
 }
 
 /* Checking the values by the end of the test */
-// setTimeout(() => {
-//   console.log(`All received data: ${array.length}`)
-// }, 5000)
-// setTimeout(() => {
-//   console.log(`All received data: ${array.length}`)
-// }, 10000)
-// setTimeout(() => {
-//   console.log(`All received data: ${array.length}`)
-// }, 15000)
-// setTimeout(() => {
-//   console.log(`All received data: ${array.length}`)
-// }, 20000)
+// function logDataLengthAfterDelay(delay) {
+//   setTimeout(() => {
+//     console.log(`All received data: ${array.length}`);
+//   }, delay);
+// }
+
+// const delays = [5000, 10000, 15000, 20000, 25000];
+
+// delays.forEach((delay) => {
+//   logDataLengthAfterDelay(delay);
+// });

test/grpc2.ts

@@ -14,7 +14,7 @@ let parsedMessages: any[] = JSON.parse(messagesJSON) // load the fake messages g
 let targetserver: string = 'localhost:3000'
 let targetserver2: string = 'localhost:3002'
 let hostServer: string = 'localhost:3001'
-let intervalToStreamOutGoingMessage: number = 1
+let intervalToStreamOutGoingMessage: number = 0.5
 let array: Message[] = []
 
 
@@ -333,16 +333,14 @@ function generateFakeStreamResponse(request: any): Subject<any> {
 }
 
 /* Checking the values by the end of the test */
-setTimeout(() => {
-  console.log(`All received data: ${array.length}`)
-}, 5000)
-setTimeout(() => {
-  console.log(`All received data: ${array.length}`)
-}, 10000)
-// setTimeout(() => {
-//   console.log(`All received data: ${array.length}`)
-// }, 15000)
-// setTimeout(() => {
-//   console.log(`All received data: ${array.length}`)
-// }, 20000)
+function logDataLengthAfterDelay(delay) {
+  setTimeout(() => {
+    console.log(`All received data: ${array.length}`);
+  }, delay);
+}
+
+const delays = [5000, 10000, 15000, 20000, 25000, 30000];
 
+delays.forEach((delay) => {
+  logDataLengthAfterDelay(delay);
+});